SEPURI-SAI-KRISHNA opened a new issue, #13022:
URL: https://github.com/apache/seatunnel/issues/13022

   ### Search before asking
   
   - [X] I had searched in the issues and found no similar issues.
   
   ### What happened
   
   `DatabendCatalog.convertDatabendType` resolves two Databend column types to 
the wrong SeaTunnel
   type. The Databend source always goes through this method: 
`DatabendSourceFactory.createSource`
   calls `catalog.getTable(tablePath)` for every job, and `getTable` calls 
`convertDatabendType` once
   per column, so there is no configuration that avoids it.
   
   **1. Every DECIMAL column resolves to STRING.**
   
   The driver reads column metadata from `information_schema.columns`, whose 
view definition is
   
   ```sql
   LOWER(data_type) AS data_type
   ```
   
   and `DatabendDatabaseMetaData.getColumns` maps `data_type as TYPE_NAME`. 
Databend builds
   `data_type` from `TableDataType::sql_name()`, which for a decimal formats as 
`Decimal(p, s)`. So a
   `DECIMAL(10, 2)` column arrives as the string `decimal(10, 2)`.
   
   The switch compares that against the bare name:
   
   ```java
   case "DECIMAL":
       return new DecimalType(columnSize, decimalDigits);
   ```
   
   `"DECIMAL(10, 2)"` never equals `"DECIMAL"`, so the branch is unreachable 
and the column falls to
   
   ```java
   default:
       log.warn("Unsupported Databend type: {}, fallback to STRING type", 
typeName);
       return BasicType.STRING_TYPE;
   ```
   
   The driver does supply the right numbers: `DatabendRawType` parses the 
precision and scale out of
   the name into `COLUMN_SIZE` and `DECIMAL_DIGITS`, and `DatabendCatalog` 
reads both. They are passed
   into a branch that cannot fire.
   
   **2. Every BINARY column resolves to TINYINT.**
   
   ```java
   case "VARBINARY":
   case "BINARY":
       return BasicType.BYTE_TYPE;
   ```
   
   `BasicType.BYTE_TYPE` is `new BasicType<>(Byte.class, SqlType.TINYINT)`, a 
single byte, not a byte
   string. Three things in the repository disagree with it:
   
   - `docs/en/connectors/source/Databend.md` and 
`docs/zh/connectors/source/Databend.md` both document
     `BINARY | BYTES`.
   - `toDatabendTypeString` in the same class writes SeaTunnel `BYTES` out as 
`VARBINARY`, so the
     forward and reverse directions do not round trip.
   - `DatabendSourceReader.convertDatabendTypeToSeaTunnelType`, the connector's 
other mapper, returns
     `PrimitiveByteArrayType.INSTANCE` for the same column.
   
   That last point is the user visible effect. 
`DatabendSource.getProducedCatalogTables()` prefers
   `reader.getRowType()` but falls back to the catalog table when `reader` is 
null, which is the case
   during job planning, before `createReader` runs. So the plan is built on the 
catalog's types while
   the reader emits values typed by its own, correct mapping: `BigDecimal` for 
a column advertised as
   STRING, and `byte[]` for a column advertised as a single `Byte`.
   
   **3. The same switch is also locale dependent.**
   
   ```java
   typeName = typeName.toUpperCase();
   ```
   
   Because the names arrive lower cased, uppercasing them with the default 
locale under `tr-TR` or
   `az-AZ` turns every `i` into a dotted capital I, and the name stops matching 
its own branch.
   Measured against the names Databend reports, 7 of 12 base types degrade to 
STRING on such a worker:
   `tinyint`, `smallint`, `int`, `bigint`, `timestamp`, `binary` and `decimal`. 
The driver itself uses
   `equalsIgnoreCase` for its own lookups and is unaffected.
   
   Here is the full mapping, produced by running the switch exactly as it 
stands against the type
   names Databend reports:
   
   ```
                      docs say       code returns (Locale.ROOT)   code returns 
(tr-TR)
   boolean            BOOLEAN        BOOLEAN                      BOOLEAN
   tinyint            TINYINT        TINYINT                      STRING   <- 
wrong
   smallint           SMALLINT       SMALLINT                     STRING   <- 
wrong
   int                INT            INT                          STRING   <- 
wrong
   bigint             BIGINT         BIGINT                       STRING   <- 
wrong
   float              FLOAT          FLOAT                        FLOAT
   double             DOUBLE         DOUBLE                       DOUBLE
   varchar            STRING         STRING                       STRING
   date               DATE           DATE                         DATE
   timestamp          TIMESTAMP      TIMESTAMP                    STRING   <- 
wrong
   binary             BYTES          TINYINT  <- wrong            STRING   <- 
wrong
   decimal(10, 2)     DECIMAL(10,2)  STRING   <- wrong            STRING   <- 
wrong
   ```
   
   ### SeaTunnel Version
   
   dev, verified at commit `eb8b80400`.
   
   ### SeaTunnel Config
   
   ```conf
   env {
     execution.parallelism = 1
     job.mode = "BATCH"
   }
   
   source {
     Databend {
       url = "jdbc:databend://databend:8000/default"
       username = "root"
       password = ""
       database = "default"
       table = "type_mapping_table"
       query = "SELECT id, price, payload FROM type_mapping_table ORDER BY id"
       fetch_size = 1000
     }
   }
   
   sink {
     Console {
     }
   }
   ```
   
   against
   
   ```sql
   CREATE TABLE type_mapping_table (id INT, price DECIMAL(10, 2), payload 
BINARY);
   INSERT INTO type_mapping_table VALUES (1, 1234.56, UNHEX('68656c6c6f'));
   ```
   
   ### Running Command
   
   ```shell
   ./bin/seatunnel.sh --config config/databend_type_mapping.conf -e local
   ```
   
   ### Error Exception
   
   ```log
   No stack trace is quoted here on purpose. The defect is in a pure type 
mapping function, and the
   evidence above is taken from the code and from running that switch directly 
rather than from a
   server run, so quoting an exception would mean inventing one.
   
   What the log does show on an affected column is only a warning, which is why 
a DECIMAL column is
   read as text without the job failing:
   
     WARN  DatabendCatalog - Unsupported Databend type: DECIMAL(10, 2), 
fallback to STRING type
   ```
   
   ### Zeta or Flink or Spark Version
   
   Not version specific. The mapping runs in the source factory, before any 
engine is involved.
   
   ### Java or Scala Version
   
   Not version specific for defects 1 and 2. Defect 3 depends on the default 
locale of the JVM running
   the job, not on its version.
   
   ### Are you willing to submit PR?
   
   - [X] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's Code of Conduct
   


-- 
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]

Reply via email to