SQL Source Connector fails to map SQL Money type to RisingWave Decimal #21761

tiphdousset · 2025-05-07T16:35:44Z

Describe the bug

SQL Source Connector fails to map SQL Money type to RisingWave Decimal type.

Error message/log

risingwave-standalone-1  | 2025-05-07T13:57:06.999433838Z  INFO            rw-streaming actor{otel.name="Actor 54" actor_id=54 prev_epoch=8479770242056192 curr_epoch=8479770279542784}:executor{otel.name="StreamCdcScan 3600002715"}: risingwave_stream::executor::backfill::cdc::cdc_backfill: start cdc backfill loop table_id=12 upstream_table_name="dbo.ProductPrices" initial_binlog_offset=Some(SqlServer(SqlServerOffset { change_lsn: "0000002c:00000ec8:0022", commit_lsn: "ffffffff:ffffffff:ffff" })) current_pk_pos=None is_snapshot_paused=false
risingwave-standalone-1  |
risingwave-standalone-1  | thread 'rw-streaming' panicked at /risingwave/src/common/src/array/mod.rs:502:29:
risingwave-standalone-1  | type mismatch, array builder type: Decimal, scalar type: Float64
risingwave-standalone-1  | stack backtrace:
risingwave-standalone-1  |    0: rust_begin_unwind
risingwave-standalone-1  |              at ./rustc/f9e0239a7bc813b4aceffc7f069f4797cde3175c/library/std/src/panicking.rs:695:5
risingwave-standalone-1  |    1: core::panicking::panic_fmt
risingwave-standalone-1  |              at ./rustc/f9e0239a7bc813b4aceffc7f069f4797cde3175c/library/core/src/panicking.rs:75:14
[...]
risingwave-standalone-1  |  149: tokio::runtime::blocking::pool::Spawner::spawn_thread::{{closure}}
risingwave-standalone-1  |              at ./root/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.40.0/src/runtime/blocking/pool.rs:476:13
risingwave-standalone-1  | note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.
risingwave-standalone-1  |
risingwave-standalone-1  |risingwave-standalone-1  | *** await tree context of current task ***
risingwave-standalone-1  |
risingwave-standalone-1  | Actor 54: `productprices` [1.425s]
risingwave-standalone-1  |   Epoch 8479770345340928 [421.000ms]
risingwave-standalone-1  |     StreamCdcScan 3600002715 [421.000ms]
risingwave-standalone-1  |       backfill_snapshot_read [421.000ms]  <== current
risingwave-standalone-1  |       Merge 360000271E [421.000ms]
risingwave-standalone-1  |         LocalInput (actor 55) [421.000ms]
risingwave-standalone-1  |

To Reproduce

Create a SQL Source connector with the Debezium parameter debezium.decimal.handling.mode = 'double'

CREATE SOURCE IF NOT EXISTS sqlserver WITH (
    connector = 'sqlserver-cdc',
    hostname = 'mssql',
    port = '1433',
    username = 'sa',
    password = 'YourStrong@Passw0rd',
    database.name = 'DemoDB',
    database.encrypt = 'true',
    debezium.tombstones.on.delete = 'false',
    debezium.decimal.handling.mode = 'double',
    debezium.time.precision.mode = 'connect',
);

Create a MSSQL Table with a column of type Money

CREATE TABLE ProductPrices (
    id INT IDENTITY(1, 1) PRIMARY KEY,
    price MONEY
);

Create a RW Table and map the SQL Money column to a RW decimal column:

CREATE table if not exists ProductPrices (
   id int,
   price decimal,
   CONSTRAINT pk_ProductPrices PRIMARY KEY (id)
) 
WITH (snapshot = 'true')
FROM sqlserver TABLE 'DemoDB.dbo.ProductPrices';

Expected behavior

I except RW to correctly handle SQL Money type and the RW table to be successfully filled in.

How did you deploy RisingWave?

I've seen the bug in both my Docker-Compose and Kubernetes environments.

My Docker-Compose is build as follows:

services:
  risingwave-standalone:
    image: risingwavelabs/risingwave:v2.3.0
    command: "standalone --meta-opts=\" \
                    --listen-addr 0.0.0.0:5690 \
                    --advertise-addr 0.0.0.0:5690 \
                    --dashboard-host 0.0.0.0:5691 \
                    --prometheus-host 0.0.0.0:1250 \
                    --prometheus-endpoint http://prometheus-0:9500 \
                    --backend sql \
                    --sql-endpoint postgres://risingwave:risingwave@postgres:5432/risingwave \
                    --state-store hummock+minio://hummockadmin:hummockadmin@minio:9301/hummock001 \
                    --data-directory hummock_001 \
                    --config-path /risingwave.toml\" \
                 --compute-opts=\" \
                    --config-path /risingwave.toml \
                    --listen-addr 0.0.0.0:5688 \
                    --prometheus-listener-addr 0.0.0.0:1250 \
                    --advertise-addr 0.0.0.0:5688 \
                    --async-stack-trace verbose \
                    --parallelism 8 \
                    --total-memory-bytes 21474836480 \
                    --role both \
                    --meta-address http://0.0.0.0:5690 \
                    --memory-manager-target-bytes 22333829939 \" \
                 --frontend-opts=\" \
                   --config-path /risingwave.toml \
                   --listen-addr 0.0.0.0:4566 \
                   --advertise-addr 0.0.0.0:4566 \
                   --prometheus-listener-addr 0.0.0.0:1250 \
                   --health-check-listener-addr 0.0.0.0:6786 \
                   --meta-addr http://0.0.0.0:5690 \
                   --frontend-total-memory-bytes=4294967296\" \
                 --compactor-opts=\" \
                   --listen-addr 0.0.0.0:6660 \
                   --prometheus-listener-addr 0.0.0.0:1250 \
                   --advertise-addr 0.0.0.0:6660 \
                   --meta-address http://0.0.0.0:5690 \
                   --compactor-total-memory-bytes=4294967296\""
    expose:
      - "6660"
      - "4566"
      - "5688"
      - "5690"
      - "1250"
      - "5691"
    ports:
      - "4566:4566"
      - "5690:5690"
      - "5691:5691"
      - "1250:1250"
    depends_on:
      - postgres
      - minio
    volumes:
      - "./risingwave.toml:/risingwave.toml"
    environment:
      RUST_BACKTRACE: "1"
      # If ENABLE_TELEMETRY is not set, telemetry will start by default
      ENABLE_TELEMETRY: ${ENABLE_TELEMETRY:-true}
      RW_TELEMETRY_TYPE: ${RW_TELEMETRY_TYPE:-"docker-compose"}
      RW_SECRET_STORE_PRIVATE_KEY_HEX: ${RW_SECRET_STORE_PRIVATE_KEY_HEX:-0123456789abcdef0123456789abcdef}
      RW_LICENSE_KEY: ${RW_LICENSE_KEY:-}
    networks:
      default:
        aliases:
          - warehouse.minio
    healthcheck:
      test:
        - CMD-SHELL
        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
        - bash -c '> /dev/tcp/127.0.0.1/4566; exit $$?;'
        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
      interval: 1s
      timeout: 5s
    restart: always
    deploy:
      resources:
        limits:
          memory: 28G
        reservations:
          memory: 28G
  postgres:
    image: "postgres:15-alpine"
    environment:
      - POSTGRES_HOST_AUTH_METHOD=trust
      - POSTGRES_USER=postgres
      - POSTGRES_DB=metadata
      - POSTGRES_INITDB_ARGS=--encoding=UTF-8 --lc-collate=C --lc-ctype=C
    expose:
      - "5432"
    ports:
      - "8432:5432"
    volumes:
      - "postgres:/var/lib/postgresql/data"
      - ./compose-resources/postgres/init-db.sql:/docker-entrypoint-initdb.d/i
8000
nit-db.sql
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready -U postgres" ]
      interval: 2s
      timeout: 5s
      retries: 5
    restart: always
  minio:
    image: minio/minio
    command:
      - server
      - "--address"
      - "0.0.0.0:9301"
      - "--console-address"
      - "0.0.0.0:9400"
      - /data
    expose:
      - "9301"
      - "9400"
    ports:
      - "9301:9301"
      - "9400:9400"
      - 9001:9001
      - 9000:9000
    depends_on: [ ]
    volumes:
      - "minio:/data"
    entrypoint: "

      /bin/sh -c '

      set -e

      mkdir -p \"/data/hummock001\"
      mkdir -p \"/data/warehouse\"

      /usr/bin/mc policy set public data/warehouse;

      /usr/bin/docker-entrypoint.sh \"$$0\" \"$$@\"

      '"
    environment:
      MINIO_CI_CD: "1"
      MINIO_PROMETHEUS_AUTH_TYPE: public
      MINIO_PROMETHEUS_URL: "http://prometheus-0:9500"
      MINIO_ROOT_PASSWORD: hummockadmin
      MINIO_ROOT_USER: hummockadmin
      MINIO_DOMAIN: "minio"
    healthcheck:
      test:
        - CMD-SHELL
        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/9301; exit $$?;'
      interval: 1s
      timeout: 5s
      retries: 5
    restart: always
    networks:
      default:
        aliases:
          - warehouse.minio
  mssql:
    build: 
      context: ./compose-resources/mssql
      dockerfile: Dockerfile
    ports:
    - "1434:1433"
    environment:
      - SA_PASSWORD=YourStrong@Passw0rd
      - ACCEPT_EULA=Y
      - MSSQL_PID=Developer
      - MSSQL_AGENT_ENABLED=true
volumes:
  postgres:
    external: false
  minio:
    external: false
  mssql-data:


### The version of RisingWave

select version();

PostgreSQL 13.14.0-RisingWave-2.3.0 (e8f053e)


### Additional context

_No response_

The text was updated successfully, but these errors were encountered:

wcy-fdu · 2025-05-08T04:33:10Z

Thanks for raising this issue, we will fix this later!

BugenZhao · 2025-05-08T05:39:14Z

There are several issues:

tiberius always maps money to double, which can be lossy: https://github.com/prisma/tiberius/blob/406ad2780d206617bd41689b1b638bddf4538f89/src/tds/codec/column_data/money.rs#L3-L24

When converting tiberius::Row to ours, we don't check the data types required by the caller at all, which leads to panicking (opened an issue at SQL Server cdc backfill panics if user-specified type mismatches with the mapping we hardcoded #21821):

risingwave/src/connector/src/parser/sql_server.rs

Lines 32 to 47 in 7a9b4d3

    
           pub fn sql_server_row_to_owned_row(row: &mut Row, schema: &Schema) -> OwnedRow { 
        
               let mut datums: Vec<Option<ScalarImpl>> = vec![]; 
        
               for i in 0..schema.fields.len() { 
        
                   let rw_field = &schema.fields[i]; 
        
                   let name = rw_field.name.as_str(); 
        
                   let datum = match row.try_get::<ScalarImplTiberiusWrapper, usize>(i) { 
        
                       Ok(datum) => datum.map(|d| d.0), 
        
                       Err(err) => { 
        
                           log_error!(name, err, "parse column failed"); 
        
                           None 
        
                       } 
        
                   }; 
        
                   datums.push(datum); 
        
               } 
        
               OwnedRow::new(datums) 
        
           }

Debezium handles incremental data while Rust code (tiberius here) handles existing (backfilling) part, which may have different type mapping. Especially that the behavior of type mapping can be changed via Debezium config, while the corresponding backfill process is not aware of this.

cc @xiangjinwu

tiphdousset added the type/bug Type: Bug. Only for issues. label May 7, 2025

github-actions bot added this to the release-2.4 milestone May 7, 2025

wcy-fdu self-assigned this May 8, 2025

BugenZhao modified the milestones: release-2.4, release-2.5 May 8, 2025

BugenZhao mentioned this issue May 12, 2025

SQL Server cdc backfill panics if user-specified type mismatches with the mapping we hardcoded #21821

Open

wcy-fdu closed this as completed May 22, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

SQL Source Connector fails to map SQL Money type to RisingWave Decimal #21761

SQL Source Connector fails to map SQL Money type to RisingWave Decimal #21761

Uh oh!

Uh oh!

SQL Source Connector fails to map SQL Money type to RisingWave Decimal #21761

SQL Source Connector fails to map SQL Money type to RisingWave Decimal #21761

Comments

Describe the bug

Error message/log

To Reproduce

Expected behavior

How did you deploy RisingWave?

Uh oh!

Uh oh!

Uh oh!