This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12862-e5e347591b5e731ae0a5c3edd61b2f2f8fa198b2 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 2cd3400d3fb73930eea7320bd68cac38c2f15fc4 Author: zhengxiang378928908-code <[email protected]> AuthorDate: Wed Oct 21 11:51:35 2026 +0000 [Fix][Translation][Spark] Carry DECIMAL with precision above 38 through Spark (#12862) Co-authored-by: Claude Opus 5.5 <[email protected]> --- .../api-design/spark-translation-layer.md | 18 + docs/en/engines/spark.md | 15 + .../introduction/concepts/incompatible-changes.md | 11 + .../api-design/spark-translation-layer.md | 18 + docs/zh/engines/spark.md | 15 + .../introduction/concepts/incompatible-changes.md | 11 + .../connector/assertion/FakeSourceToAssertIT.java | 26 + .../assertion/fake_wide_decimal_map_to_assert.conf | 55 +++ .../assertion/fake_wide_decimal_to_assert.conf | 146 ++++++ .../spark/exception/SparkTranslationErrorCode.java | 50 ++ .../spark/execution/MultiTableManager.java | 7 + .../spark/serialization/InternalRowConverter.java | 63 ++- .../spark/serialization/SeaTunnelRowConverter.java | 74 ++- .../spark/utils/TypeConverterUtils.java | 40 +- .../translation/spark/utils/WideDecimalUtils.java | 285 +++++++++++ .../translation/spark/utils/WideDecimalTest.java | 536 +++++++++++++++++++++ 16 files changed, 1348 insertions(+), 22 deletions(-) diff --git a/docs/en/architecture/api-design/spark-translation-layer.md b/docs/en/architecture/api-design/spark-translation-layer.md index 6b75b29202..72503e0bba 100644 --- a/docs/en/architecture/api-design/spark-translation-layer.md +++ b/docs/en/architecture/api-design/spark-translation-layer.md @@ -102,6 +102,24 @@ This is one of the most sensitive boundaries in the Spark path, especially for: - nested types - nullability +### DECIMAL precision above 38 + +Spark's `DecimalType` stops at precision 38. `TypeConverterUtils` maps a SeaTunnel `DecimalType(p, s)` with `p > 38` as follows, in the same way `TIME` and `TIMESTAMP_TZ` are carried with a metadata flag: + +| Position of the wide decimal | Spark schema | +|---|---| +| Top-level field, or a field of a `ROW` when every container on the path is a `ROW` | `StringType`, with `StructField` metadata `seatunnel_decimal_precision = p` and `seatunnel_decimal_scale = s` (both long) | +| Anywhere inside an `ARRAY` or `MAP` | rejected with `SPARK-TRANSLATION-01` | + +Rules: + +- `MultiTableManager` checks every source `CatalogTable` before `mergeSchema`, because merging renames the columns to `column0..N`. The error names the table, the column path (`.element`, `.key`, `.value`), the declared type, the limit of 38 and the workarounds. +- Write (`InternalRowConverter` and `SeaTunnelRowConverter`): `null` stays `null`. The value is padded with `setScale(s, RoundingMode.UNNECESSARY)` and written with `toPlainString()`. A value that would need rounding, or has more than `p` digits, is rejected with `SPARK-TRANSLATION-02`. +- Read: the text must match `-?\d+(\.\d+)?`, have exactly `s` fraction digits and at most `p` digits. Anything else is rejected with `SPARK-TRANSLATION-02`, never rescaled. +- Schema read-back (`TypeConverterUtils.convert(StructType)`): a `StringType` field with both metadata keys becomes `DecimalType(p, s)` again. Only one key, the keys on another type, or `p <= 38` is rejected with `SPARK-TRANSLATION-03`. + +Row values are converted from the SeaTunnel row type, so transforms and sinks always see `DecimalType(p, s)`. + ## Version Split Spark translation in SeaTunnel is version-aware. Spark 2.4 and Spark 3.x do not expose identical datasource APIs, so SeaTunnel keeps separate modules and adapters for major Spark lines. diff --git a/docs/en/engines/spark.md b/docs/en/engines/spark.md index 1a045b8650..acfcfff910 100644 --- a/docs/en/engines/spark.md +++ b/docs/en/engines/spark.md @@ -98,6 +98,21 @@ sink { If you need more transform options, see [Transforms Catalog](../transforms) and [Transform Common Options](../transforms/common-options/common-options.md). +## DECIMAL Precision Above 38 + +Spark's `DECIMAL` type supports a precision of at most 38. SeaTunnel types can be wider, for example a Doris `LARGEINT` (`DECIMAL(39, 0)`) or a PostgreSQL `NUMERIC(50, 10)`. + +- **Top-level columns and fields inside a `ROW`**: SeaTunnel carries the value through Spark as plain decimal text in a `STRING` column, and records the declared precision and scale in the column metadata. Transforms and sinks still receive `DECIMAL(p, s)` with the exact value, so auto-created sink tables get the same column type as on the Zeta and Flink engines. Only Spark's own view of the data, such as the Dataset schema or a temporary view, shows the column as `STRING`. +- **Inside an `ARRAY` or `MAP`** (element, key or value, at any depth): there is no column to hold the metadata, so the job fails at start, before any data is read. The error names the table, the column path (for example `tags.element` or `attrs.value`), the declared type and the Spark limit. + +SeaTunnel never rounds or truncates these values. A value that does not fit the declared precision or scale fails the job with an error naming the column. + +Workarounds for an `ARRAY` or `MAP` column. They all act at the source, because on Spark the source schema is built before any transform runs: + +- leave the column out at the source, for example with Doris `doris.read.field` or a JDBC `query` that omits it; +- cast it to a string, or to a narrower type if you know the values fit, in the source query; +- run the job on the Zeta or Flink engine, which keep `DECIMAL(p > 38)`. + ## Running From A Source Checkout If you are running examples from the repository source tree, the example module is: diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index eff49771be..b24119253f 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -5,6 +5,17 @@ You need to check this document before you upgrade to related version. ## dev +### Spark: DECIMAL with precision above 38 + +- **Behavior change: on the Spark engine, `DECIMAL(p, s)` with `p > 38` no longer fails at schema creation when it is a top-level column or a field inside a `ROW`** + - **Description**: Spark's decimal type stops at precision 38, so any such column failed with `decimal can only support precision up to 38` before any row was read. SeaTunnel now carries these values through Spark as plain decimal text, with the declared precision and scale in the column metadata, and restores `DECIMAL(p, s)` for transforms and sinks. + - **Impact**: + - Jobs that read such columns on Spark 2.4 or 3.3 now run and keep every digit. + - A `DECIMAL(p > 38)` inside an `ARRAY` or `MAP` now fails at job start with error `SPARK-TRANSLATION-01`, which names the table, the column path and the workarounds. If the Doris source declares `LARGEINT` as `DECIMAL(39, 0)`, this applies to Doris `ARRAY<LARGEINT>` columns and `LARGEINT` map keys and values on Spark, which could overflow above 20 digits with the old `DECIMAL(20, 0)` type. + - A value that does not fit the declared precision or scale fails the job with `SPARK-TRANSLATION-02` instead of being rounded. + - In Spark's own view of the data, such as the Dataset schema or a temporary view, these columns are `STRING`. + - **Migration Guide**: No change is needed for top-level or `ROW` columns. For an `ARRAY` or `MAP` column, leave it out at the source, cast it in the source query, or run the job on the Zeta or Flink engine. A `Sql` transform cannot help here, because on Spark the source schema is built before any transform runs. See [SeaTunnel With Spark](../../engines/spark.md#decimal-precision-above-38). + ### File sink: XML applies `date_format`, `datetime_format` and `time_format` - **Behavior change: the XML writer of the file sinks formats DATE, TIME and TIMESTAMP values with the configured formats** diff --git a/docs/zh/architecture/api-design/spark-translation-layer.md b/docs/zh/architecture/api-design/spark-translation-layer.md index 3eafd70af1..ff2ec292f7 100644 --- a/docs/zh/architecture/api-design/spark-translation-layer.md +++ b/docs/zh/architecture/api-design/spark-translation-layer.md @@ -102,6 +102,24 @@ Spark 转换层对 schema 转换非常敏感，因为 Spark 本身依赖严格 - 嵌套类型 - nullability +### 精度超过 38 的 DECIMAL + +Spark 的 `DecimalType` 精度最大为 38。`TypeConverterUtils` 对 `p > 38` 的 SeaTunnel `DecimalType(p, s)` 按下表映射，做法与 `TIME`、`TIMESTAMP_TZ` 借助 metadata 标记传递相同： + +| 宽 DECIMAL 的位置 | Spark schema | +|---|---| +| 顶层字段，或路径上所有容器都是 `ROW` 时 `ROW` 内的字段 | `StringType`，`StructField` metadata 为 `seatunnel_decimal_precision = p` 和 `seatunnel_decimal_scale = s`（均为 long） | +| `ARRAY` 或 `MAP` 内任意位置 | 以 `SPARK-TRANSLATION-01` 拒绝 | + +规则： + +- `MultiTableManager` 在 `mergeSchema` 之前检查每个 Source 的 `CatalogTable`，因为合并后列名会变成 `column0..N`。报错信息包含表名、列路径（`.element`、`.key`、`.value`）、声明的类型、上限 38 以及绕过办法。 +- 写入（`InternalRowConverter` 和 `SeaTunnelRowConverter`）：`null` 保持为 `null`。值先用 `setScale(s, RoundingMode.UNNECESSARY)` 补齐小数位，再用 `toPlainString()` 输出。需要四舍五入或位数超过 `p` 的值以 `SPARK-TRANSLATION-02` 拒绝。 +- 读取：文本必须匹配 `-?\d+(\.\d+)?`，小数位数必须正好是 `s`，总位数不超过 `p`。其他情况一律以 `SPARK-TRANSLATION-02` 拒绝，不会重新调整小数位数。 +- Schema 回读（`TypeConverterUtils.convert(StructType)`）：带有两个 metadata 键的 `StringType` 字段会还原为 `DecimalType(p, s)`。只有一个键、键出现在其他类型上或 `p <= 38` 时，以 `SPARK-TRANSLATION-03` 拒绝。 + +行数据按 SeaTunnel 的行类型转换，因此 Transform 和 Sink 看到的始终是 `DecimalType(p, s)`。 + ## 版本拆分 SeaTunnel 的 Spark 转换层是按版本线拆分的。Spark 2.4 和 Spark 3.x 的 datasource API 不完全一致，因此仓库里保留了不同 major 版本的独立适配模块。 diff --git a/docs/zh/engines/spark.md b/docs/zh/engines/spark.md index 53f1ef160d..f2a3878717 100644 --- a/docs/zh/engines/spark.md +++ b/docs/zh/engines/spark.md @@ -98,6 +98,21 @@ sink { 如果你需要更多 transform 能力，继续查看 [Transforms 目录](../transforms) 和 [Transform 通用参数](../transforms/common-options/common-options.md)。 +## 精度超过 38 的 DECIMAL + +Spark 的 `DECIMAL` 类型精度最大为 38，而 SeaTunnel 的类型可以更宽，例如 Doris `LARGEINT`（`DECIMAL(39, 0)`）或 PostgreSQL `NUMERIC(50, 10)`。 + +- **顶层列和 `ROW` 内的字段**：SeaTunnel 在 Spark 中把值作为普通十进制文本放在 `STRING` 列里传递，并在列的 metadata 中记录声明的精度和小数位数。Transform 和 Sink 收到的仍然是 `DECIMAL(p, s)` 和原值，所以 Sink 自动建表得到的列类型与 Zeta、Flink 引擎一致。只有 Spark 自己看到的数据（例如 Dataset 的 schema 或临时视图）中，该列显示为 `STRING`。 +- **在 `ARRAY` 或 `MAP` 内**（元素、键或值，任意嵌套层级）：没有可以存放 metadata 的列，因此作业在启动时、读取任何数据之前就会失败。报错信息包含表名、列路径（例如 `tags.element` 或 `attrs.value`）、声明的类型以及 Spark 的上限。 + +SeaTunnel 不会对这些值做四舍五入或截断。值超出声明的精度或小数位数时，作业会失败，并在报错中给出列名。 + +`ARRAY` 或 `MAP` 列的绕过办法。这些办法都在 Source 端生效，因为在 Spark 上 Source 的 schema 在任何 Transform 运行之前就已构建： + +- 在 Source 端不读取该列，例如使用 Doris 的 `doris.read.field`，或在 JDBC 的 `query` 中不选该列； +- 在 Source 查询中把它转换为字符串；如果确定数值放得下，也可以转换为更窄的类型； +- 改用 Zeta 或 Flink 引擎运行作业，这两个引擎支持 `DECIMAL(p > 38)`。 + ## 从源码仓库运行示例 如果你是在源码仓库里运行示例，对应模块是： diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index bb0974bc39..f2c287a693 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -4,6 +4,17 @@ ## dev +### Spark：精度超过 38 的 DECIMAL + +- **行为变更：在 Spark 引擎上，作为顶层列或 `ROW` 内字段的 `DECIMAL(p, s)`（`p > 38`）不再在构建 schema 时失败** + - **描述**：Spark 的 decimal 类型精度最大为 38，因此此前这类列会在读取任何数据之前报错 `decimal can only support precision up to 38`。现在 SeaTunnel 在 Spark 中把这些值作为普通十进制文本传递，在列的 metadata 中记录声明的精度和小数位数，并为 Transform 和 Sink 还原为 `DECIMAL(p, s)`。 + - **影响**： + - 在 Spark 2.4 或 3.3 上读取这类列的作业现在可以运行，且不丢失任何数字。 + - `ARRAY` 或 `MAP` 内的 `DECIMAL(p > 38)` 现在会在作业启动时以错误 `SPARK-TRANSLATION-01` 失败，报错中给出表名、列路径和绕过办法。如果 Doris 源把 `LARGEINT` 声明为 `DECIMAL(39, 0)`，这也适用于 Spark 上 Doris 的 `ARRAY<LARGEINT>` 列以及 MAP 中的 `LARGEINT` 键和值；此前它们使用 `DECIMAL(20, 0)`，超过 20 位的值会溢出。 + - 值超出声明的精度或小数位数时，作业会以 `SPARK-TRANSLATION-02` 失败，不会被四舍五入。 + - 在 Spark 自己看到的数据中（例如 Dataset 的 schema 或临时视图），这些列是 `STRING`。 + - **迁移指南**：顶层列和 `ROW` 内的列无需修改。对于 `ARRAY` 或 `MAP` 列，请在 Source 端不读取该列、在 Source 查询中转换它，或改用 Zeta 或 Flink 引擎运行作业。`Sql` Transform 在这里无效，因为在 Spark 上 Source 的 schema 在任何 Transform 运行之前就已构建。参见 [SeaTunnel 运行在 Spark 上](../../engines/spark.md#精度超过-38-的-decimal)。 + ### File Sink：XML 应用 `date_format`、`datetime_format` 和 `time_format` - **行为变更：文件类 Sink 的 XML 写入器按配置的格式输出 DATE、TIME 和 TIMESTAMP** diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/java/org/apache/seatunnel/e2e/connector/assertion/FakeSourceToAssertIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/java/org/apache/seatunnel/e2e/connector/assertion/FakeSourceToAssertIT.java index f366f69b6e..70c2c00f29 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/java/org/apache/seatunnel/e2e/connector/assertion/FakeSourceToAssertIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/java/org/apache/seatunnel/e2e/connector/assertion/FakeSourceToAssertIT.java @@ -69,4 +69,30 @@ public class FakeSourceToAssertIT extends TestSuiteBase { container.executeJob("/assertion/fake_full_types_to_assert.conf"); Assertions.assertEquals(0, execResult.getExitCode()); } + + /** DECIMAL above precision 38 keeps its values and type, also through a transform on Spark. */ + @TestTemplate + public void testWideDecimalToAssertSink(TestContainer container) + throws IOException, InterruptedException { + Container.ExecResult execResult = + container.executeJob("/assertion/fake_wide_decimal_to_assert.conf"); + Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); + } + + /** Spark cannot carry DECIMAL above precision 38 inside a MAP and must say so at job start. */ + @TestTemplate + public void testWideDecimalInMapToAssertSink(TestContainer container) + throws IOException, InterruptedException { + Container.ExecResult execResult = + container.executeJob("/assertion/fake_wide_decimal_map_to_assert.conf"); + if (container.identifier().getEngineType() != EngineType.SPARK) { + Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); + return; + } + Assertions.assertNotEquals(0, execResult.getExitCode()); + String output = execResult.getStdout() + execResult.getStderr(); + Assertions.assertTrue(output.contains("SPARK-TRANSLATION-01"), output); + Assertions.assertTrue(output.contains("`c_attrs.value`"), output); + Assertions.assertTrue(output.contains("DECIMAL(39, 0)"), output); + } } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fake_wide_decimal_map_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fake_wide_decimal_map_to_assert.conf new file mode 100644 index 0000000000..47e8b0f608 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fake_wide_decimal_map_to_assert.conf @@ -0,0 +1,55 @@ +# +# 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. +# + +# DECIMAL with precision above 38 as a MAP value. Spark rejects this at job start. + +env { + parallelism = 1 + job.mode = BATCH +} + +source { + FakeSource { + schema = { + fields { + c_id = int + c_attrs = "map<string, decimal(39, 0)>" + } + } + rows = [ + { + kind = INSERT + fields = [1, { k0 = "170141183460469231731687303715884105727" }] + } + ] + plugin_output = "fake" + } +} + +sink { + Assert { + plugin_input = "fake" + rules = { + row_rules = [ + { + rule_type = MIN_ROW + rule_value = 1 + } + ] + } + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fake_wide_decimal_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fake_wide_decimal_to_assert.conf new file mode 100644 index 0000000000..9a58bebef6 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fake_wide_decimal_to_assert.conf @@ -0,0 +1,146 @@ +# +# 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. +# + +# DECIMAL with precision above 38, at the top level and inside a ROW. +# Spark carries these as text. The sink must still get BigDecimal values with the +# declared scale (checked by field_type) and the exact digits (checked by equals_to). + +env { + parallelism = 1 + job.mode = BATCH +} + +source { + FakeSource { + schema = { + fields { + c_id = int + c_max = "decimal(39, 0)" + c_min = "decimal(39, 0)" + c_zero = "decimal(39, 0)" + c_null = "decimal(39, 0)" + c_scaled = "decimal(50, 10)" + c_narrow = "decimal(10, 2)" + c_row = { + c_inner_max = "decimal(39, 0)" + c_inner_min = "decimal(39, 0)" + } + } + } + rows = [ + { + kind = INSERT + fields = [1, "170141183460469231731687303715884105727", "-170141183460469231731687303715884105728", "0", null, "-12345678901234567890123456789012345678.0123456789", "12345678.90", ["170141183460469231731687303715884105727", "-170141183460469231731687303715884105728"]] + } + ] + plugin_output = "fake" + } +} + +transform { + Sql { + plugin_input = "fake" + plugin_output = "fake1" + query = "select c_id, c_max, c_min, c_zero, c_null, c_scaled, c_narrow, c_row from dual" + } +} + +sink { + Assert { + plugin_input = "fake1" + rules = { + row_rules = [ + { + rule_type = MAX_ROW + rule_value = 1 + }, + { + rule_type = MIN_ROW + rule_value = 1 + } + ] + field_rules = [ + { + field_name = c_max + field_type = "decimal(39, 0)" + field_value = [ + { + equals_to = "170141183460469231731687303715884105727" + } + ] + }, + { + field_name = c_min + field_type = "decimal(39, 0)" + field_value = [ + { + equals_to = "-170141183460469231731687303715884105728" + } + ] + }, + { + field_name = c_zero + field_type = "decimal(39, 0)" + field_value = [ + { + equals_to = "0" + } + ] + }, + { + field_name = c_null + field_type = "decimal(39, 0)" + field_value = [ + { + rule_type = NULL + } + ] + }, + { + field_name = c_scaled + field_type = "decimal(50, 10)" + field_value = [ + { + equals_to = "-12345678901234567890123456789012345678.0123456789" + } + ] + }, + { + field_name = c_narrow + field_type = "decimal(10, 2)" + field_value = [ + { + equals_to = "12345678.90" + } + ] + }, + { + field_name = c_row + field_type = { + c_inner_max = "decimal(39, 0)" + c_inner_min = "decimal(39, 0)" + } + field_value = [ + { + equals_to = ["170141183460469231731687303715884105727", "-170141183460469231731687303715884105728"] + } + ] + } + ] + } + } +} diff --git a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/exception/SparkTranslationErrorCode.java b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/exception/SparkTranslationErrorCode.java new file mode 100644 index 0000000000..3473de6041 --- /dev/null +++ b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/exception/SparkTranslationErrorCode.java @@ -0,0 +1,50 @@ +/* + * 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.translation.spark.exception; + +import org.apache.seatunnel.common.exception.SeaTunnelErrorCode; + +/** Error codes of the Spark translation layer. */ +public enum SparkTranslationErrorCode implements SeaTunnelErrorCode { + WIDE_DECIMAL_IN_COLLECTION( + "SPARK-TRANSLATION-01", + "DECIMAL with precision above 38 inside an ARRAY or MAP is not supported on Spark"), + INVALID_WIDE_DECIMAL_VALUE( + "SPARK-TRANSLATION-02", + "Value does not match the declared DECIMAL type with precision above 38"), + INVALID_WIDE_DECIMAL_METADATA( + "SPARK-TRANSLATION-03", "Invalid SeaTunnel decimal metadata on a Spark schema field"); + + private final String code; + private final String description; + + SparkTranslationErrorCode(String code, String description) { + this.code = code; + this.description = description; + } + + @Override + public String getCode() { + return code; + } + + @Override + public String getDescription() { + return description; + } +} diff --git a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/execution/MultiTableManager.java b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/execution/MultiTableManager.java index 48b7d766bf..2d1dd5d933 100644 --- a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/execution/MultiTableManager.java +++ b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/execution/MultiTableManager.java @@ -27,6 +27,7 @@ import org.apache.seatunnel.translation.spark.serialization.InternalRowCollector import org.apache.seatunnel.translation.spark.serialization.InternalRowConverter; import org.apache.seatunnel.translation.spark.serialization.SeaTunnelRowConverter; import org.apache.seatunnel.translation.spark.utils.TypeConverterUtils; +import org.apache.seatunnel.translation.spark.utils.WideDecimalUtils; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.GenericRow; @@ -58,6 +59,12 @@ public class MultiTableManager implements Serializable { private boolean isMultiTable = false; public MultiTableManager(CatalogTable[] catalogTables) { + // Merging renames the columns to column0..N, so check before it to name the real column. + for (CatalogTable catalogTable : catalogTables) { + WideDecimalUtils.checkNoWideDecimalInCollection( + String.valueOf(catalogTable.getTablePath()), + catalogTable.getSeaTunnelRowType()); + } List<ColumnWithIndex> columnWithIndexes = mergeSchema(catalogTables); if (catalogTables.length > 1) { isMultiTable = true; diff --git a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/InternalRowConverter.java b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/InternalRowConverter.java index 0258c52661..bacf29c758 100644 --- a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/InternalRowConverter.java +++ b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/InternalRowConverter.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.translation.spark.serialization; import org.apache.seatunnel.api.table.type.ArrayType; import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; import org.apache.seatunnel.api.table.type.MapType; import org.apache.seatunnel.api.table.type.RowKind; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; @@ -28,6 +29,7 @@ import org.apache.seatunnel.translation.serialization.RowConverter; import org.apache.seatunnel.translation.spark.utils.InstantConverterUtils; import org.apache.seatunnel.translation.spark.utils.OffsetDateTimeUtils; import org.apache.seatunnel.translation.spark.utils.TypeConverterUtils; +import org.apache.seatunnel.translation.spark.utils.WideDecimalUtils; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.MutableAny; @@ -90,6 +92,14 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { } private static Object convert(Object field, SeaTunnelDataType<?> dataType) { + return convert(field, dataType, null); + } + + /** + * @param column the field name, used in error messages; {@code null} inside an ARRAY or MAP. A + * nested ROW adds its own name only when an error passes through it. + */ + private static Object convert(Object field, SeaTunnelDataType<?> dataType, String column) { if (field == null) { return null; } @@ -97,7 +107,11 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { case ROW: SeaTunnelRow seaTunnelRow = (SeaTunnelRow) field; SeaTunnelRowType rowType = (SeaTunnelRowType) dataType; - return convert(seaTunnelRow, rowType); + try { + return convert(seaTunnelRow, rowType); + } catch (WideDecimalUtils.InvalidWideDecimalException e) { + throw e.withParent(column); + } case DATE: return (int) ((LocalDate) field).toEpochDay(); case TIME: @@ -112,6 +126,11 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { case STRING: return UTF8String.fromString((String) field); case DECIMAL: + if (WideDecimalUtils.isWideDecimal(dataType)) { + return UTF8String.fromString( + WideDecimalUtils.toCarrierText( + (BigDecimal) field, (DecimalType) dataType, column)); + } return Decimal.apply((BigDecimal) field); case ARRAY: SeaTunnelDataType<?> elementType = ((ArrayType<?, ?>) dataType).getElementType(); @@ -155,7 +174,11 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { if (TypeConverterUtils.ROW_KIND_FIELD.equals(rowType.getFieldName(i))) { values[i].update(seaTunnelRow.getRowKind().toByteValue()); } else { - Object fieldValue = convert(seaTunnelRow.getField(i), rowType.getFieldType(i)); + Object fieldValue = + convert( + seaTunnelRow.getField(i), + rowType.getFieldType(i), + rowType.getFieldName(i)); if (fieldValue != null) { values[i].update(fieldValue); } @@ -170,7 +193,11 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { MutableValue[] values = new MutableValue[arity + 2]; for (int i = 0; i < indexes.length; i++) { values[indexes[i] + 2] = createMutableValue(rowType.getFieldType(indexes[i])); - Object fieldValue = convert(seaTunnelRow.getField(i), rowType.getFieldType(indexes[i])); + Object fieldValue = + convert( + seaTunnelRow.getField(i), + rowType.getFieldType(indexes[i]), + rowType.getFieldName(indexes[i])); if (fieldValue != null) { values[indexes[i] + 2].update(fieldValue); } @@ -275,7 +302,8 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { engineRow.get( indexes[i] + 2, TypeConverterUtils.convert(rowType.getFieldType(indexes[i]))), - rowType.getFieldType(indexes[i])); + rowType.getFieldType(indexes[i]), + rowType.getFieldName(indexes[i])); } SeaTunnelRow seaTunnelRow = new SeaTunnelRow(fields); seaTunnelRow.setRowKind(rowKind); @@ -289,12 +317,24 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { } private static Object reconvert(Object field, SeaTunnelDataType<?> dataType) { + return reconvert(field, dataType, null); + } + + /** + * @param column the field name, used in error messages; {@code null} inside an ARRAY or MAP. A + * nested ROW adds its own name only when an error passes through it. + */ + private static Object reconvert(Object field, SeaTunnelDataType<?> dataType, String column) { if (field == null) { return null; } switch (dataType.getSqlType()) { case ROW: - return reconvert((InternalRow) field, (SeaTunnelRowType) dataType); + try { + return reconvert((InternalRow) field, (SeaTunnelRowType) dataType); + } catch (WideDecimalUtils.InvalidWideDecimalException e) { + throw e.withParent(column); + } case DATE: if (field instanceof Date) { return ((Date) field).toLocalDate(); @@ -333,11 +373,19 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { case STRING: return field.toString(); case DECIMAL: - if (field instanceof Decimal) { + if (WideDecimalUtils.isWideDecimal(dataType) + && (field instanceof UTF8String || field instanceof String)) { + return WideDecimalUtils.fromCarrierText( + field.toString(), (DecimalType) dataType, column); + } else if (field instanceof Decimal) { return ((Decimal) field).toJavaBigDecimal(); } else if (field instanceof BigDecimal) { return field; } + throw new IllegalArgumentException( + String.format( + "Unexpected Spark value of type %s for SeaTunnel %s", + field.getClass().getName(), dataType)); case ARRAY: if (field instanceof ArrayData) { return reconvertArray((ArrayData) field, (ArrayType<?, ?>) dataType); @@ -361,7 +409,8 @@ public final class InternalRowConverter extends RowConverter<InternalRow> { fields[i] = reconvert( engineRow.get(i, TypeConverterUtils.convert(rowType.getFieldType(i))), - rowType.getFieldType(i)); + rowType.getFieldType(i), + rowType.getFieldName(i)); } return new SeaTunnelRow(fields); } diff --git a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java index 1d03b6448d..36b8ddae85 100644 --- a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java +++ b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.translation.spark.serialization; import org.apache.seatunnel.api.table.type.ArrayType; import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; import org.apache.seatunnel.api.table.type.MapType; import org.apache.seatunnel.api.table.type.RowKind; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; @@ -27,6 +28,7 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.api.table.type.SqlType; import org.apache.seatunnel.translation.serialization.RowConverter; import org.apache.seatunnel.translation.spark.utils.OffsetDateTimeUtils; +import org.apache.seatunnel.translation.spark.utils.WideDecimalUtils; import org.apache.spark.sql.catalyst.expressions.GenericRow; import org.apache.spark.unsafe.types.UTF8String; @@ -76,7 +78,11 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { fields[0] = seaTunnelRow.getRowKind().toByteValue(); fields[1] = seaTunnelRow.getTableId(); for (int i = 0; i < indexes.length; i++) { - Object fieldValue = convert(seaTunnelRow.getField(i), rowType.getFieldType(indexes[i])); + Object fieldValue = + convert( + seaTunnelRow.getField(i), + rowType.getFieldType(indexes[i]), + rowType.getFieldName(indexes[i])); if (fieldValue != null) { fields[indexes[i] + 2] = fieldValue; } @@ -85,6 +91,14 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { } private Object convert(Object field, SeaTunnelDataType<?> dataType) { + return convert(field, dataType, null); + } + + /** + * @param column the field name, used in error messages; {@code null} inside an ARRAY or MAP. A + * nested ROW adds its own name only when an error passes through it. + */ + private Object convert(Object field, SeaTunnelDataType<?> dataType, String column) { if (field == null) { return null; } @@ -92,7 +106,17 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { case ROW: SeaTunnelRow seaTunnelRow = (SeaTunnelRow) field; SeaTunnelRowType rowType = (SeaTunnelRowType) dataType; - return convertRow(seaTunnelRow, rowType); + try { + return convertRow(seaTunnelRow, rowType); + } catch (WideDecimalUtils.InvalidWideDecimalException e) { + throw e.withParent(column); + } + case DECIMAL: + if (WideDecimalUtils.isWideDecimal(dataType)) { + return WideDecimalUtils.toCarrierText( + (BigDecimal) field, (DecimalType) dataType, column); + } + return field; case DATE: return Date.valueOf((LocalDate) field); case TIMESTAMP: @@ -138,7 +162,11 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { int arity = rowType.getTotalFields(); Object[] values = new Object[arity]; for (int i = 0; i < arity; i++) { - Object fieldValue = convert(seaTunnelRow.getField(i), rowType.getFieldType(i)); + Object fieldValue = + convert( + seaTunnelRow.getField(i), + rowType.getFieldType(i), + rowType.getFieldName(i)); if (fieldValue != null) { values[i] = fieldValue; } @@ -197,7 +225,11 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { Object[] fields = new Object[indexes.length]; for (int i = 0; i < indexes.length; i++) { int fieldIndex = indexes[i]; - fields[i] = reconvert(engineRow.get(fieldIndex + 2), rowType.getFieldType(fieldIndex)); + fields[i] = + reconvert( + engineRow.get(fieldIndex + 2), + rowType.getFieldType(fieldIndex), + rowType.getFieldName(fieldIndex)); } SeaTunnelRow seaTunnelRow = new SeaTunnelRow(fields); seaTunnelRow.setRowKind(rowKind); @@ -206,15 +238,35 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { } private Object reconvert(Object field, SeaTunnelDataType<?> dataType) { + return reconvert(field, dataType, null); + } + + /** + * @param column the field name, used in error messages; {@code null} inside an ARRAY or MAP. A + * nested ROW adds its own name only when an error passes through it. + */ + private Object reconvert(Object field, SeaTunnelDataType<?> dataType, String column) { if (field == null) { return null; } switch (dataType.getSqlType()) { case ROW: - if (field instanceof GenericRow) { - return createFromGenericRow((GenericRow) field, (SeaTunnelRowType) dataType); + try { + if (field instanceof GenericRow) { + return createFromGenericRow( + (GenericRow) field, (SeaTunnelRowType) dataType); + } + return reconvert((SeaTunnelRow) field, (SeaTunnelRowType) dataType); + } catch (WideDecimalUtils.InvalidWideDecimalException e) { + throw e.withParent(column); + } + case DECIMAL: + if (WideDecimalUtils.isWideDecimal(dataType) + && (field instanceof String || field instanceof UTF8String)) { + return WideDecimalUtils.fromCarrierText( + field.toString(), (DecimalType) dataType, column); } - return reconvert((SeaTunnelRow) field, (SeaTunnelRowType) dataType); + return field; case DATE: return ((Date) field).toLocalDate(); case TIMESTAMP: @@ -241,7 +293,7 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { Object[] fields = row.values(); Object[] newFields = new Object[fields.length]; for (int idx = 0; idx < fields.length; idx++) { - newFields[idx] = reconvert(fields[idx], type.getFieldType(idx)); + newFields[idx] = reconvert(fields[idx], type.getFieldType(idx), type.getFieldName(idx)); } return new SeaTunnelRow(newFields); } @@ -250,7 +302,11 @@ public class SeaTunnelRowConverter extends RowConverter<GenericRow> { int num = engineRow.getFields().length; Object[] fields = new Object[num]; for (int i = 0; i < num; i++) { - fields[i] = reconvert(engineRow.getFields()[i], rowType.getFieldType(i)); + fields[i] = + reconvert( + engineRow.getFields()[i], + rowType.getFieldType(i), + rowType.getFieldName(i)); } return new SeaTunnelRow(fields); } diff --git a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/utils/TypeConverterUtils.java b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/utils/TypeConverterUtils.java index 45ead72e6a..13a601c216 100644 --- a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/utils/TypeConverterUtils.java +++ b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/utils/TypeConverterUtils.java @@ -68,6 +68,16 @@ public class TypeConverterUtils { } public static DataType convert(SeaTunnelDataType<?> dataType) { + return convert(dataType, null); + } + + /** + * @param collectionPart {@code "element"}, {@code "key"} or {@code "value"} inside an ARRAY or + * MAP, else {@code null}. A wide decimal there fails. {@code MultiTableManager} already + * rejects it with the table and column at job start; this only keeps other callers safe + * without walking the type tree again. + */ + private static DataType convert(SeaTunnelDataType<?> dataType, String collectionPart) { checkNotNull(dataType, "The SeaTunnel's data type is required."); switch (dataType.getSqlType()) { case NULL: @@ -100,23 +110,32 @@ public class TypeConverterUtils { return OffsetDateTimeUtils.OFFSET_DATETIME_WITH_DECIMAL; case ARRAY: return DataTypes.createArrayType( - convert(((ArrayType<?, ?>) dataType).getElementType())); + convert(((ArrayType<?, ?>) dataType).getElementType(), "element")); case MAP: MapType<?, ?> mapType = (MapType<?, ?>) dataType; return DataTypes.createMapType( - convert(mapType.getKeyType()), convert(mapType.getValueType())); + convert(mapType.getKeyType(), "key"), + convert(mapType.getValueType(), "value")); case DECIMAL: + if (WideDecimalUtils.isWideDecimal(dataType)) { + if (collectionPart != null) { + throw WideDecimalUtils.collectionError( + null, collectionPart, (DecimalType) dataType); + } + // Spark stops at precision 38; the field metadata keeps the declared type. + return DataTypes.StringType; + } DecimalType decimalType = (DecimalType) dataType; return new org.apache.spark.sql.types.DecimalType( decimalType.getPrecision(), decimalType.getScale()); case ROW: - return convert((SeaTunnelRowType) dataType); + return convert((SeaTunnelRowType) dataType, collectionPart); default: } throw new IllegalArgumentException("Unsupported SeaTunnel's data type: " + dataType); } - private static StructType convert(SeaTunnelRowType rowType) { + private static StructType convert(SeaTunnelRowType rowType, String collectionPart) { // TODO: row kind StructField[] fields = new StructField[rowType.getFieldNames().length]; for (int i = 0; i < rowType.getFieldNames().length; i++) { @@ -131,12 +150,18 @@ public class TypeConverterUtils { OffsetDateTimeUtils.LOGICAL_TIMESTAMP_WITH_OFFSET_TYPE_FLAG, true) .build(); + } else if (WideDecimalUtils.isWideDecimal(fieldType)) { + metadata = WideDecimalUtils.metadata((DecimalType) fieldType); } else { metadata = Metadata.empty(); } fields[i] = - new StructField(rowType.getFieldNames()[i], convert(fieldType), true, metadata); + new StructField( + rowType.getFieldNames()[i], + convert(fieldType, collectionPart), + true, + metadata); } return new StructType(fields); } @@ -210,7 +235,10 @@ public class TypeConverterUtils { for (int i = 0; i < structFields.length; i++) { fieldNames[i] = structFields[i].name(); Metadata metadata = structFields[i].metadata(); - if (metadata != null + DecimalType wideDecimalType = WideDecimalUtils.fromMetadata(structFields[i]); + if (wideDecimalType != null) { + fieldTypes[i] = wideDecimalType; + } else if (metadata != null && metadata.contains(LOGICAL_TIME_TYPE_FLAG) && metadata.getBoolean(LOGICAL_TIME_TYPE_FLAG)) { fieldTypes[i] = LocalTimeType.LOCAL_TIME_TYPE; diff --git a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/utils/WideDecimalUtils.java b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/utils/WideDecimalUtils.java new file mode 100644 index 0000000000..0175f91755 --- /dev/null +++ b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/utils/WideDecimalUtils.java @@ -0,0 +1,285 @@ +/* + * 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.translation.spark.utils; + +import org.apache.seatunnel.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; +import org.apache.seatunnel.translation.spark.exception.SparkTranslationErrorCode; + +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.MetadataBuilder; +import org.apache.spark.sql.types.StructField; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.util.regex.Pattern; + +/** + * Carries SeaTunnel {@code DECIMAL(p, s)} with {@code p > 38} through Spark, whose decimal type + * stops at precision 38. + * + * <p>A wide decimal at the top level or inside a {@code ROW} becomes a Spark {@code StringType} + * field with the declared precision and scale in the field metadata. The value is its plain decimal + * text. A wide decimal inside an {@code ARRAY} or {@code MAP} has no field to hold the metadata, so + * it is rejected before the job reads any data. + */ +public final class WideDecimalUtils { + + public static final int SPARK_MAX_PRECISION = 38; + public static final String PRECISION_KEY = "seatunnel_decimal_precision"; + public static final String SCALE_KEY = "seatunnel_decimal_scale"; + + private static final Pattern PLAIN_DECIMAL = Pattern.compile("-?\\d+(\\.\\d+)?"); + private static final int MAX_VALUE_LENGTH_IN_MESSAGE = 64; + + private WideDecimalUtils() { + throw new UnsupportedOperationException( + "WideDecimalUtils is a utility class and cannot be instantiated"); + } + + public static boolean isWideDecimal(SeaTunnelDataType<?> dataType) { + return dataType instanceof DecimalType + && ((DecimalType) dataType).getPrecision() > SPARK_MAX_PRECISION; + } + + public static Metadata metadata(DecimalType decimalType) { + return new MetadataBuilder() + .putLong(PRECISION_KEY, decimalType.getPrecision()) + .putLong(SCALE_KEY, decimalType.getScale()) + .build(); + } + + /** + * Restores the SeaTunnel type of a carrier field. + * + * @return the decimal type, or {@code null} if the field carries no decimal metadata + */ + public static DecimalType fromMetadata(StructField field) { + Metadata metadata = field.metadata(); + boolean hasPrecision = metadata != null && metadata.contains(PRECISION_KEY); + boolean hasScale = metadata != null && metadata.contains(SCALE_KEY); + if (!hasPrecision && !hasScale) { + return null; + } + if (!hasPrecision || !hasScale) { + throw invalidMetadata(field, "both keys must be present"); + } + if (!DataTypes.StringType.equals(field.dataType())) { + throw invalidMetadata(field, "the field type must be STRING"); + } + long precision; + long scale; + try { + precision = metadata.getLong(PRECISION_KEY); + scale = metadata.getLong(SCALE_KEY); + } catch (RuntimeException e) { + throw invalidMetadata(field, "both values must be integers"); + } + if (precision <= SPARK_MAX_PRECISION + || precision > Integer.MAX_VALUE + || scale < 0 + || scale > precision) { + throw invalidMetadata( + field, + String.format( + "precision must be above %d and scale between 0 and precision", + SPARK_MAX_PRECISION)); + } + return new DecimalType((int) precision, (int) scale); + } + + /** + * Fails if a wide decimal sits inside an {@code ARRAY} or {@code MAP} anywhere in the row type. + * Call it once per source table, before the tables are merged, so that the error can name the + * real table and column. + * + * @param tablePath the table path for the error message, or {@code null} if unknown + */ + public static void checkNoWideDecimalInCollection(String tablePath, SeaTunnelRowType rowType) { + for (int i = 0; i < rowType.getTotalFields(); i++) { + check(tablePath, rowType.getFieldName(i), rowType.getFieldType(i), false); + } + } + + private static void check( + String tablePath, String path, SeaTunnelDataType<?> dataType, boolean inCollection) { + switch (dataType.getSqlType()) { + case DECIMAL: + if (inCollection && isWideDecimal(dataType)) { + throw collectionError(tablePath, path, (DecimalType) dataType); + } + return; + case ROW: + SeaTunnelRowType rowType = (SeaTunnelRowType) dataType; + for (int i = 0; i < rowType.getTotalFields(); i++) { + check( + tablePath, + path + "." + rowType.getFieldName(i), + rowType.getFieldType(i), + inCollection); + } + return; + case ARRAY: + check( + tablePath, + path + ".element", + ((ArrayType<?, ?>) dataType).getElementType(), + true); + return; + case MAP: + MapType<?, ?> mapType = (MapType<?, ?>) dataType; + check(tablePath, path + ".key", mapType.getKeyType(), true); + check(tablePath, path + ".value", mapType.getValueType(), true); + return; + default: + } + } + + /** SeaTunnel value to carrier text. Pads the scale but never rounds or narrows. */ + public static String toCarrierText(BigDecimal value, DecimalType decimalType, String column) { + BigDecimal scaled; + try { + scaled = value.setScale(decimalType.getScale(), RoundingMode.UNNECESSARY); + } catch (ArithmeticException e) { + throw invalidValue( + column, + decimalType, + value.toPlainString(), + "it has more than " + decimalType.getScale() + " fraction digits"); + } + if (scaled.precision() > decimalType.getPrecision()) { + throw invalidValue( + column, + decimalType, + value.toPlainString(), + "it has more than " + decimalType.getPrecision() + " digits"); + } + return scaled.toPlainString(); + } + + /** Carrier text to SeaTunnel value. Requires the exact declared scale and never rescales. */ + public static BigDecimal fromCarrierText(String text, DecimalType decimalType, String column) { + if (!PLAIN_DECIMAL.matcher(text).matches()) { + throw invalidValue(column, decimalType, text, "it is not plain decimal text"); + } + BigDecimal value = new BigDecimal(text); + if (value.scale() != decimalType.getScale()) { + throw invalidValue( + column, + decimalType, + text, + "its scale is " + value.scale() + ", expected " + decimalType.getScale()); + } + if (value.precision() > decimalType.getPrecision()) { + throw invalidValue( + column, + decimalType, + text, + "it has more than " + decimalType.getPrecision() + " digits"); + } + return value; + } + + static String typeName(DecimalType decimalType) { + return String.format("DECIMAL(%d, %d)", decimalType.getPrecision(), decimalType.getScale()); + } + + static SeaTunnelRuntimeException collectionError( + String tablePath, String columnPath, DecimalType decimalType) { + String table = tablePath == null ? "" : String.format(" of table `%s`", tablePath); + return new SeaTunnelRuntimeException( + SparkTranslationErrorCode.WIDE_DECIMAL_IN_COLLECTION, + String.format( + "Column `%s`%s has type %s inside an ARRAY/MAP. " + + "Spark supports DECIMAL precision up to %d, " + + "and SeaTunnel cannot carry wider decimals inside ARRAY/MAP on Spark. " + + "Workarounds: leave the column out at the source " + + "(e.g. Doris `doris.read.field`, or a JDBC `query` that omits it), " + + "cast it to a string in the source query, " + + "or run this job on the Zeta or Flink engine.", + columnPath, table, typeName(decimalType), SPARK_MAX_PRECISION)); + } + + private static InvalidWideDecimalException invalidValue( + String column, DecimalType decimalType, String value, String reason) { + return new InvalidWideDecimalException(column, decimalType, abbreviate(value), reason); + } + + private static SeaTunnelRuntimeException invalidMetadata(StructField field, String reason) { + return new SeaTunnelRuntimeException( + SparkTranslationErrorCode.INVALID_WIDE_DECIMAL_METADATA, + String.format( + "Field `%s` of type %s has metadata %s, but %s.", + field.name(), field.dataType().sql(), field.metadata().json(), reason)); + } + + /** + * A wide decimal value that cannot be carried. The converters pass only the field name down and + * each enclosing {@code ROW} adds its name on the way up, so the column path is built only when + * a value fails. + */ + public static final class InvalidWideDecimalException extends SeaTunnelRuntimeException { + + private static final long serialVersionUID = 1L; + + private final String column; + private final DecimalType decimalType; + private final String value; + private final String reason; + + private InvalidWideDecimalException( + String column, DecimalType decimalType, String value, String reason) { + super( + SparkTranslationErrorCode.INVALID_WIDE_DECIMAL_VALUE, + String.format( + "Column `%s` declared as %s: value '%s' is rejected because %s.", + column, typeName(decimalType), value, reason)); + this.column = column; + this.decimalType = decimalType; + this.value = value; + this.reason = reason; + } + + /** @return the same error with {@code parent} prepended to the column path */ + public InvalidWideDecimalException withParent(String parent) { + if (parent == null) { + return this; + } + InvalidWideDecimalException e = + new InvalidWideDecimalException( + column == null ? parent : parent + "." + column, + decimalType, + value, + reason); + e.setStackTrace(getStackTrace()); + return e; + } + } + + private static String abbreviate(String value) { + if (value.length() <= MAX_VALUE_LENGTH_IN_MESSAGE) { + return value; + } + return value.substring(0, MAX_VALUE_LENGTH_IN_MESSAGE) + "..."; + } +} diff --git a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/test/java/org/apache/seatunnel/translation/spark/utils/WideDecimalTest.java b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/test/java/org/apache/seatunnel/translation/spark/utils/WideDecimalTest.java new file mode 100644 index 0000000000..fadfd2b42d --- /dev/null +++ b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/test/java/org/apache/seatunnel/translation/spark/utils/WideDecimalTest.java @@ -0,0 +1,536 @@ +/* + * 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.translation.spark.utils; + +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; +import org.apache.seatunnel.translation.spark.exception.SparkTranslationErrorCode; +import org.apache.seatunnel.translation.spark.execution.MultiTableManager; +import org.apache.seatunnel.translation.spark.serialization.InternalRowConverter; +import org.apache.seatunnel.translation.spark.serialization.SeaTunnelRowConverter; + +import org.apache.spark.sql.Row; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer$; +import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder; +import org.apache.spark.sql.catalyst.encoders.RowEncoder; +import org.apache.spark.sql.catalyst.expressions.GenericRow; +import org.apache.spark.sql.catalyst.expressions.MutableAny; +import org.apache.spark.sql.catalyst.expressions.MutableByte; +import org.apache.spark.sql.catalyst.expressions.MutableValue; +import org.apache.spark.sql.catalyst.expressions.SpecificInternalRow; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.MetadataBuilder; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.unsafe.types.UTF8String; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.function.Executable; + +import java.io.IOException; +import java.math.BigDecimal; +import java.util.Arrays; +import java.util.List; + +class WideDecimalTest { + + private static final BigDecimal INT128_MIN = + new BigDecimal("-170141183460469231731687303715884105728"); + private static final BigDecimal INT128_MAX = + new BigDecimal("170141183460469231731687303715884105727"); + private static final BigDecimal SCALED = + new BigDecimal("-12345678901234567890123456789012345678.0123456789"); + + private static final DecimalType DECIMAL_39_0 = new DecimalType(39, 0); + private static final DecimalType DECIMAL_50_10 = new DecimalType(50, 10); + private static final DecimalType DECIMAL_10_2 = new DecimalType(10, 2); + + private static final SeaTunnelRowType INFO_TYPE = + new SeaTunnelRowType( + new String[] {"inner_big", "name"}, + new SeaTunnelDataType<?>[] {DECIMAL_39_0, BasicType.STRING_TYPE}); + + /** Wide decimals at the top level and inside a ROW, next to an ordinary decimal. */ + private static final SeaTunnelRowType ROW_TYPE = + new SeaTunnelRowType( + new String[] {"id", "big", "scaled", "narrow", "info"}, + new SeaTunnelDataType<?>[] { + BasicType.INT_TYPE, DECIMAL_39_0, DECIMAL_50_10, DECIMAL_10_2, INFO_TYPE + }); + + private static List<SeaTunnelRow> sampleRows() { + return Arrays.asList( + row(1, INT128_MAX, SCALED, new BigDecimal("12345678.90"), INT128_MIN), + row(2, INT128_MIN, SCALED.negate(), new BigDecimal("-0.01"), INT128_MAX), + row(3, BigDecimal.ZERO, new BigDecimal("0E-10"), BigDecimal.ZERO.setScale(2), null), + row(4, null, null, null, null)); + } + + private static SeaTunnelRow row( + int id, BigDecimal big, BigDecimal scaled, BigDecimal narrow, BigDecimal innerBig) { + SeaTunnelRow info = new SeaTunnelRow(new Object[] {innerBig, "n" + id}); + SeaTunnelRow row = new SeaTunnelRow(new Object[] {id, big, scaled, narrow, info}); + row.setTableId("db.t"); + return row; + } + + @Test + void schemaUsesStringCarrierWithMetadata() { + StructType schema = (StructType) TypeConverterUtils.convert(ROW_TYPE); + + assertCarrier(schema.apply("big"), 39, 0); + assertCarrier(schema.apply("scaled"), 50, 10); + Assertions.assertEquals( + new org.apache.spark.sql.types.DecimalType(10, 2), + schema.apply("narrow").dataType()); + Assertions.assertEquals(Metadata.empty(), schema.apply("narrow").metadata()); + + StructType info = (StructType) schema.apply("info").dataType(); + assertCarrier(info.apply("inner_big"), 39, 0); + } + + @Test + void schemaRoundTripKeepsDecimalType() { + DataType sparkType = TypeConverterUtils.convert(ROW_TYPE); + Assertions.assertEquals(ROW_TYPE, TypeConverterUtils.convert(sparkType)); + } + + @Test + void parcelledSchemaKeepsCarrier() { + StructType schema = (StructType) TypeConverterUtils.parcel(ROW_TYPE); + assertCarrier(schema.apply("big"), 39, 0); + } + + @Test + void internalRowConverterRoundTrip() throws IOException { + InternalRowConverter converter = new InternalRowConverter(ROW_TYPE); + for (SeaTunnelRow row : sampleRows()) { + InternalRow internalRow = converter.convert(row); + Object big = internalRow.get(3, DataTypes.StringType); + if (row.getField(1) == null) { + Assertions.assertTrue(internalRow.isNullAt(3)); + } else { + Assertions.assertEquals( + UTF8String.fromString(((BigDecimal) row.getField(1)).toPlainString()), big); + } + assertRowEquals(row, converter.reconvert(internalRow)); + } + } + + @Test + void seaTunnelRowConverterRoundTripThroughRowEncoder() throws IOException { + SeaTunnelRowConverter converter = new SeaTunnelRowConverter(ROW_TYPE); + StructType schema = (StructType) TypeConverterUtils.parcel(ROW_TYPE); + ExpressionEncoder<Row> encoder = rowEncoder(schema); + for (SeaTunnelRow row : sampleRows()) { + GenericRow genericRow = converter.convert(row); + if (row.getField(1) != null) { + Assertions.assertEquals( + ((BigDecimal) row.getField(1)).toPlainString(), genericRow.get(3)); + } + // The transform path hands these rows to Spark through a RowEncoder. + InternalRow encoded = encoder.createSerializer().apply(genericRow).copy(); + Row decoded = encoder.createDeserializer().apply(encoded); + assertRowEquals(row, converter.reconvert((GenericRow) decoded)); + } + } + + @Test + void writePadsScaleWithoutRounding() throws IOException { + SeaTunnelRow row = row(5, new BigDecimal("7"), new BigDecimal("5"), null, null); + InternalRowConverter converter = new InternalRowConverter(ROW_TYPE); + InternalRow internalRow = converter.convert(row); + Assertions.assertEquals( + UTF8String.fromString("5.0000000000"), internalRow.get(4, DataTypes.StringType)); + Assertions.assertEquals( + new BigDecimal("5.0000000000"), converter.reconvert(internalRow).getField(2)); + } + + @Test + void writeRejectsRoundingAndOverflow() { + InternalRowConverter converter = new InternalRowConverter(ROW_TYPE); + assertInvalidValue( + () -> converter.convert(row(6, new BigDecimal("1.5"), BigDecimal.ONE, null, null)), + "`big`", + "DECIMAL(39, 0)", + "1.5"); + BigDecimal fortyDigits = new BigDecimal("1" + repeat('0', 39)); + assertInvalidValue( + () -> converter.convert(row(7, fortyDigits, BigDecimal.ONE, null, null)), + "`big`", + "DECIMAL(39, 0)", + "more than 39 digits"); + SeaTunnelRowConverter genericConverter = new SeaTunnelRowConverter(ROW_TYPE); + assertInvalidValue( + () -> + genericConverter.convert( + row(8, BigDecimal.ONE, BigDecimal.ONE, null, fortyDigits)), + "`info.inner_big`", + "DECIMAL(39, 0)", + "more than 39 digits"); + } + + @Test + void readRejectsMalformedCarrierText() { + for (String text : + new String[] {"", " 1", "1 ", "1e5", "1E+5", "NaN", "+1", "1,000", "0x1F", "-"}) { + assertInvalidValue( + () -> WideDecimalUtils.fromCarrierText(text, DECIMAL_39_0, "big"), + "`big`", + "DECIMAL(39, 0)", + "not plain decimal text"); + } + assertInvalidValue( + () -> WideDecimalUtils.fromCarrierText("1.5", DECIMAL_39_0, "big"), + "`big`", + "DECIMAL(39, 0)", + "scale is 1, expected 0"); + assertInvalidValue( + () -> WideDecimalUtils.fromCarrierText("1.00", DECIMAL_50_10, "scaled"), + "`scaled`", + "DECIMAL(50, 10)", + "scale is 2, expected 10"); + assertInvalidValue( + () -> WideDecimalUtils.fromCarrierText("1" + repeat('0', 39), DECIMAL_39_0, "big"), + "`big`", + "DECIMAL(39, 0)", + "more than 39 digits"); + } + + @Test + void readErrorNamesNestedColumnAndTruncatesValue() { + String longText = "x" + repeat('1', 100); + GenericRow info = new GenericRow(new Object[] {longText, "n"}); + GenericRow row = new GenericRow(new Object[] {(byte) 0, "db.t", 1, null, null, null, info}); + SeaTunnelRuntimeException e = + assertInvalidValue( + () -> new SeaTunnelRowConverter(ROW_TYPE).reconvert(row), + "`info.inner_big`", + "DECIMAL(39, 0)", + "not plain decimal text"); + Assertions.assertTrue(e.getMessage().contains(longText.substring(0, 64) + "...")); + Assertions.assertFalse(e.getMessage().contains(longText)); + } + + @Test + void internalRowReadRejectsMalformedCarrierText() { + MutableValue[] values = new MutableValue[ROW_TYPE.getTotalFields() + 2]; + for (int i = 0; i < values.length; i++) { + values[i] = new MutableAny(); + } + values[0] = new MutableByte(); + values[0].update((byte) 0); + values[1].update(UTF8String.fromString("db.t")); + values[3].update(UTF8String.fromString("12.0")); + assertInvalidValue( + () -> new InternalRowConverter(ROW_TYPE).reconvert(new SpecificInternalRow(values)), + "`big`", + "DECIMAL(39, 0)", + "scale is 1, expected 0"); + } + + @Test + void ordinaryDecimalWithUnexpectedValueFailsInsteadOfFallingThrough() { + MutableValue[] values = new MutableValue[ROW_TYPE.getTotalFields() + 2]; + for (int i = 0; i < values.length; i++) { + values[i] = new MutableAny(); + } + values[0] = new MutableByte(); + values[0].update((byte) 0); + values[1].update(UTF8String.fromString("db.t")); + values[5].update("not a decimal"); + IllegalArgumentException e = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + new InternalRowConverter(ROW_TYPE) + .reconvert(new SpecificInternalRow(values))); + Assertions.assertTrue(e.getMessage().contains("java.lang.String"), e.getMessage()); + } + + @Test + void preflightRejectsWideDecimalInArray() { + assertCollectionError(arrayOf(DECIMAL_39_0), "demo.t_wide", "tags.element"); + } + + @Test + void preflightRejectsWideDecimalMapValue() { + assertCollectionError( + new MapType<>(BasicType.STRING_TYPE, DECIMAL_39_0), "demo.t_wide", "col.value"); + } + + @Test + void preflightRejectsWideDecimalMapKey() { + assertCollectionError( + new MapType<>(DECIMAL_39_0, BasicType.STRING_TYPE), "demo.t_wide", "col.key"); + } + + @Test + void preflightRejectsWideDecimalInArrayInsideRow() { + SeaTunnelRowType info = + new SeaTunnelRowType( + new String[] {"scores"}, + new SeaTunnelDataType<?>[] {arrayOf(DECIMAL_39_0)}); + assertCollectionError(info, "demo.t_wide", "col.scores.element"); + } + + @Test + void preflightAllowsNarrowDecimalInCollections() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"tags", "attrs"}, + new SeaTunnelDataType<?>[] { + arrayOf(new DecimalType(38, 0)), + new MapType<>(BasicType.STRING_TYPE, new DecimalType(38, 10)) + }); + Assertions.assertDoesNotThrow( + () -> new MultiTableManager(new CatalogTable[] {table("demo.ok", rowType)})); + } + + @Test + void preflightNamesOriginalTableAndColumnInMultiTableJob() { + SeaTunnelRowType good = + new SeaTunnelRowType( + new String[] {"id", "big"}, + new SeaTunnelDataType<?>[] {BasicType.INT_TYPE, DECIMAL_39_0}); + SeaTunnelRowType bad = + new SeaTunnelRowType( + new String[] {"id", "largeint_values"}, + new SeaTunnelDataType<?>[] {BasicType.INT_TYPE, arrayOf(DECIMAL_39_0)}); + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> + new MultiTableManager( + new CatalogTable[] { + table("demo.a_good", good), table("demo.b_bad", bad) + })); + Assertions.assertEquals( + SparkTranslationErrorCode.WIDE_DECIMAL_IN_COLLECTION, e.getSeaTunnelErrorCode()); + Assertions.assertTrue(e.getMessage().contains("demo.b_bad"), e.getMessage()); + Assertions.assertTrue(e.getMessage().contains("`largeint_values.element`"), e.getMessage()); + Assertions.assertFalse(e.getMessage().contains("column0"), e.getMessage()); + Assertions.assertFalse(e.getMessage().contains("column1"), e.getMessage()); + } + + @Test + void multiTableJobWithTopLevelWideDecimalStillRoundTrips() throws IOException { + SeaTunnelRowType a = + new SeaTunnelRowType( + new String[] {"id", "big"}, + new SeaTunnelDataType<?>[] {BasicType.INT_TYPE, DECIMAL_39_0}); + SeaTunnelRowType b = + new SeaTunnelRowType( + new String[] {"scaled"}, new SeaTunnelDataType<?>[] {DECIMAL_50_10}); + MultiTableManager manager = + new MultiTableManager(new CatalogTable[] {table("demo.a", a), table("demo.b", b)}); + StructType schema = manager.getTableSchema(); + assertCarrier(schema.apply("column1"), 39, 0); + assertCarrier(schema.apply("column2"), 50, 10); + + SeaTunnelRow rowA = new SeaTunnelRow(new Object[] {1, INT128_MIN}); + rowA.setTableId("demo.a"); + SeaTunnelRow rowB = new SeaTunnelRow(new Object[] {SCALED}); + rowB.setTableId("demo.b"); + for (SeaTunnelRow row : Arrays.asList(rowA, rowB)) { + GenericRow genericRow = manager.convert(row); + assertRowEquals(row, manager.reconvert(genericRow)); + } + } + + @Test + void collectionConversionOutsidePreflightStillFails() { + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> TypeConverterUtils.convert(arrayOf(DECIMAL_39_0))); + Assertions.assertTrue(e.getMessage().contains("`element`"), e.getMessage()); + SeaTunnelRowType rowInArray = + new SeaTunnelRowType(new String[] {"v"}, new SeaTunnelDataType<?>[] {DECIMAL_39_0}); + e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> + TypeConverterUtils.convert( + new MapType<>( + BasicType.STRING_TYPE, + new ArrayType<>( + SeaTunnelRow[].class, rowInArray)))); + Assertions.assertEquals( + SparkTranslationErrorCode.WIDE_DECIMAL_IN_COLLECTION, e.getSeaTunnelErrorCode()); + } + + @Test + void valueErrorNamesFullPathOfDeeplyNestedColumn() { + SeaTunnelRowType outerType = + new SeaTunnelRowType( + new String[] {"id", "outer"}, + new SeaTunnelDataType<?>[] { + BasicType.INT_TYPE, + new SeaTunnelRowType( + new String[] {"info"}, new SeaTunnelDataType<?>[] {INFO_TYPE}) + }); + BigDecimal fortyDigits = new BigDecimal("1" + repeat('0', 39)); + SeaTunnelRow info = new SeaTunnelRow(new Object[] {fortyDigits, "n"}); + SeaTunnelRow row = + new SeaTunnelRow(new Object[] {1, new SeaTunnelRow(new Object[] {info})}); + row.setTableId("db.t"); + assertInvalidValue( + () -> new InternalRowConverter(outerType).convert(row), + "`outer.info.inner_big`", + "more than 39 digits"); + assertInvalidValue( + () -> new SeaTunnelRowConverter(outerType).convert(row), + "`outer.info.inner_big`", + "more than 39 digits"); + } + + @Test + void metadataReadBackRejectsInconsistentFields() { + Metadata onlyPrecision = + new MetadataBuilder().putLong(WideDecimalUtils.PRECISION_KEY, 39).build(); + Metadata both = WideDecimalUtils.metadata(DECIMAL_39_0); + Metadata narrow = + new MetadataBuilder() + .putLong(WideDecimalUtils.PRECISION_KEY, 20) + .putLong(WideDecimalUtils.SCALE_KEY, 0) + .build(); + Metadata wrongType = + new MetadataBuilder() + .putString(WideDecimalUtils.PRECISION_KEY, "39") + .putLong(WideDecimalUtils.SCALE_KEY, 0) + .build(); + for (StructField field : + new StructField[] { + new StructField("a", DataTypes.StringType, true, onlyPrecision), + new StructField("b", DataTypes.LongType, true, both), + new StructField("c", DataTypes.StringType, true, narrow), + new StructField("d", DataTypes.StringType, true, wrongType) + }) { + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> + TypeConverterUtils.convert( + new StructType(new StructField[] {field}))); + Assertions.assertEquals( + SparkTranslationErrorCode.INVALID_WIDE_DECIMAL_METADATA, + e.getSeaTunnelErrorCode()); + Assertions.assertTrue(e.getMessage().contains("`" + field.name() + "`")); + } + } + + private static void assertCollectionError( + SeaTunnelDataType<?> columnType, String tablePath, String columnPath) { + String columnName = columnPath.substring(0, columnPath.indexOf('.')); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", columnName}, + new SeaTunnelDataType<?>[] {BasicType.INT_TYPE, columnType}); + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> + new MultiTableManager( + new CatalogTable[] {table(tablePath, rowType)})); + Assertions.assertEquals( + SparkTranslationErrorCode.WIDE_DECIMAL_IN_COLLECTION, e.getSeaTunnelErrorCode()); + String message = e.getMessage(); + for (String expected : + new String[] { + "`" + columnPath + "`", + tablePath, + "DECIMAL(39, 0)", + "Spark supports DECIMAL precision up to 38", + "doris.read.field", + "source query", + "Zeta or Flink" + }) { + Assertions.assertTrue(message.contains(expected), expected + " not in: " + message); + } + } + + private static SeaTunnelRuntimeException assertInvalidValue( + Executable executable, String... expectedParts) { + SeaTunnelRuntimeException e = + Assertions.assertThrows(SeaTunnelRuntimeException.class, executable); + Assertions.assertEquals( + SparkTranslationErrorCode.INVALID_WIDE_DECIMAL_VALUE, e.getSeaTunnelErrorCode()); + for (String expected : expectedParts) { + Assertions.assertTrue( + e.getMessage().contains(expected), expected + " not in: " + e.getMessage()); + } + return e; + } + + private static void assertCarrier(StructField field, long precision, long scale) { + Assertions.assertEquals(DataTypes.StringType, field.dataType(), field.name()); + Assertions.assertEquals( + precision, field.metadata().getLong(WideDecimalUtils.PRECISION_KEY), field.name()); + Assertions.assertEquals( + scale, field.metadata().getLong(WideDecimalUtils.SCALE_KEY), field.name()); + } + + private static void assertRowEquals(SeaTunnelRow expected, SeaTunnelRow actual) { + Assertions.assertEquals(expected.getArity(), actual.getArity()); + for (int i = 0; i < expected.getArity(); i++) { + Object e = expected.getField(i); + Object a = actual.getField(i); + if (e instanceof SeaTunnelRow) { + assertRowEquals((SeaTunnelRow) e, (SeaTunnelRow) a); + } else { + // BigDecimal.equals also compares the scale. + Assertions.assertEquals(e, a, "field " + i); + } + } + } + + private static ArrayType<BigDecimal[], BigDecimal> arrayOf(DecimalType type) { + return new ArrayType<>(BigDecimal[].class, type); + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private static ExpressionEncoder<Row> rowEncoder(StructType schema) { + // Spark 3.3 API; the Spark 2.4 and 3.3 end-to-end tests cover both versions. + scala.collection.Seq attributes = (scala.collection.Seq) schema.toAttributes(); + return RowEncoder.apply(schema).resolveAndBind(attributes, SimpleAnalyzer$.MODULE$); + } + + private static CatalogTable table(String path, SeaTunnelRowType rowType) { + String[] parts = path.split("\\."); + return CatalogTableUtil.getCatalogTable("test", parts[0], null, parts[1], rowType); + } + + private static String repeat(char c, int count) { + char[] chars = new char[count]; + Arrays.fill(chars, c); + return new String(chars); + } +}
