zhengxiang378928908-code opened a new issue, #12814:
URL: https://github.com/apache/seatunnel/issues/12814
### Search before asking
- [x] I searched in the issues and found nothing similar.
### Background
While fixing #12753 (Doris `LARGEINT` declared as `DECIMAL(20, 0)`), #12787
changes the Doris source type to `DECIMAL(39, 0)` so that non-Doris sinks get a
column wide enough for the full signed 128-bit range. On Zeta and Flink this
works. On Spark 2.4 / 3.3 every read of a Doris table with a `LARGEINT` column
now fails before any row is read:
```text
org.apache.spark.sql.AnalysisException: decimal can only support precision
up to 38
at org.apache.spark.sql.types.DecimalType.<init>
at
org.apache.seatunnel.translation.spark.utils.TypeConverterUtils.convert(TypeConverterUtils.java:111)
at ...MultiTableManager.getTableSchema
at ...BatchSourceReader.readSchema
```
(fork CI: `doris-connector-it` (11) and (17), Spark containers of `DorisIT`,
`DorisMultiReadIT`, `DorisLargeintToMysqlIT`.)
As discussed in #12753, this Spark limitation should not be hidden by
disabling Spark tests or by silently clamping to 38 digits, and an implicit
fallback should not be added inside #12787. This issue is the place to agree on
the Spark policy separately.
### The problem is not Doris-specific
`TypeConverterUtils.convert` maps a SeaTunnel `DecimalType(p, s)` directly
to `new org.apache.spark.sql.types.DecimalType(p, s)`, and Spark rejects `p >
38`. Any source that declares a precision above 38 hits this. From reading the
current `dev` code (not yet reproduced end to end):
| Source | Column | SeaTunnel type today | Spark |
|---|---|---|---|
| PostgreSQL | `NUMERIC(50, 0)` | `DECIMAL(50, 0)` (`PostgresTypeConverter`,
passed through) | expected to fail the same way |
| MySQL | `DECIMAL(65, 0)` | clamped to `DECIMAL(38, 18)` with a WARN "will
probably cause value overflow" (`MySqlTypeConverter`) | runs, but silently
loses range |
| Doris (#12787) | `LARGEINT` | `DECIMAL(39, 0)` | fails |
So today the project already has both behaviors that #12753 rejected: an
unexplained failure (PostgreSQL) and a silent clamp (MySQL). A Spark-level
policy would apply to all of them, not only Doris.
### Code paths that any policy has to cover
1. Schema: `TypeConverterUtils.convert(SeaTunnelDataType)` builds the Spark
`StructType` (`MultiTableManager.getTableSchema`,
`BatchSourceReader.readSchema`).
2. Row write (SeaTunnel -> Spark): `InternalRowConverter.convert`, `case
DECIMAL: Decimal.apply((BigDecimal) field)`.
3. Row read (Spark -> SeaTunnel): `InternalRowConverter.reconvert`, `case
DECIMAL`, plus the `TypeConverterUtils.convert(...)` calls used to read array
elements, map keys/values and nested rows (`InternalRowConverter.java:213-214,
277, 363, 376`).
4. Schema read-back (Spark -> SeaTunnel):
`TypeConverterUtils.convert(StructType)`, which is what the sink and transforms
see.
5. Nested types: `ARRAY<DECIMAL(39,0)>` and `MAP<..., DECIMAL(39,0)>` (Doris
`ARRAY<largeint>` / `MAP<..., largeint>` produce these).
### Options
**A. Documented string representation, opt-in or always**
Map `DecimalType(p > 38)` to Spark `StringType`; write
`BigDecimal.toPlainString()`, and the sink receives a SeaTunnel `STRING`.
- Pros: never loses digits; simple.
- Cons: the sink sees `STRING`, so auto-created tables get `VARCHAR`/`TEXT`
instead of a numeric column. That is a cross-connector schema change on Spark
only, so Spark and Zeta/Flink would produce different downstream DDL for the
same job.
**B. Clear preflight failure**
Keep the current mapping, but check during schema conversion and throw a
`SeaTunnelRuntimeException` naming the table, the column, the declared
precision, the engine limit, and the workaround (for example a `Sql` transform
`CAST(col AS STRING)` or `CAST(col AS DECIMAL(38, 0))`).
- Pros: no hidden behavior; small change; makes the PostgreSQL case
understandable too.
- Cons: Spark users of Doris `LARGEINT` who ran fine on `DECIMAL(20, 0)` now
have to add a transform. That is still a regression for them, so on its own it
is probably not enough for #12787.
**C. Preserve the SeaTunnel schema across the Spark boundary (string carrier
+ metadata)**
There is already a precedent in the same class: `TIME` and `TIMESTAMP_TZ`
are carried through Spark as a physical type plus a `StructField` metadata flag
(`LOGICAL_TIME_TYPE_FLAG`,
`OffsetDateTimeUtils.LOGICAL_TIMESTAMP_WITH_OFFSET_TYPE_FLAG`), and
`TypeConverterUtils.convert(StructType)` restores the original SeaTunnel type
from that flag. The same approach would be:
- schema: `DECIMAL(p > 38, s)` -> Spark `StringType`, with metadata such as
`seatunnel_decimal_precision = p`, `seatunnel_decimal_scale = s`;
- row write: `UTF8String.fromString(bigDecimal.toPlainString())`;
- row read: `new BigDecimal(str)`;
- schema read-back: metadata present -> `DecimalType(p, s)` again.
- Pros: the sink sees `DECIMAL(39, 0)` on every engine, so downstream DDL is
the same on Spark, Zeta and Flink; no digits are lost; follows an existing
pattern in this module.
- Cons:
- `StructField` metadata only exists on top-level fields. For
`ARRAY<DECIMAL(39,0)>` / `MAP<..., DECIMAL(39,0)>` / nested `ROW` the type
cannot be restored this way, so nested cases still need B (clear failure) or an
extra encoding.
- A Spark-side `Sql` transform on that column sees a string. That has to
be documented.
- Needs Spark 2.4 and 3.3 unit tests for schema and row round trip, plus
an E2E.
### Proposal for discussion
My current lean is **C for top-level fields + B for nested types**, because
it keeps the downstream schema identical across engines and reuses the
`TIMESTAMP_TZ` pattern. I'm happy to go with A or B if maintainers prefer a
smaller contract change.
Whichever is chosen, the change would come as its own PR with:
- Spark 2.4 / 3.3 unit tests for schema conversion and row round trip
(top-level and nested), with the extreme values `-2^127` and `2^127 - 1`;
- EN/ZH docs describing the Spark behavior for `DECIMAL` precision > 38;
- an entry in `incompatible-changes.md` if the observable schema changes on
Spark;
- the existing `DorisIT` / `DorisMultiReadIT` Spark containers kept enabled.
#12787 will link here and stay focused on the Doris source representation
and Zeta/Flink coverage until this is agreed.
Questions for maintainers:
1. Which of A / B / C (or another approach) should be the Spark contract?
2. Should the existing MySQL silent clamp to `DECIMAL(38, 18)` be in scope
here, or tracked separately?
### Related
- #12753 Doris LARGEINT declared as DECIMAL(20,0)
- #12787 Declare LARGEINT as DECIMAL(39, 0)
### Are you willing to submit a PR?
- [x] Yes I am willing to submit a PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
