From cf3d0bba2e64869f9683f5e4ac0df1abed64310e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=98=BF=E4=B8=99?=
 <50567478+gaopeng666@users.noreply.github.com>
Date: Fri, 14 Jul 2023 11:07:46 +0800
Subject: [PATCH] [Bugfix]fix clickhouse source connector read Nullable() type
 is not null,example:Nullable(Float64) while value is null the result is 0.0
 (#5080)

---
 .../clickhouse/source/ClickhouseSourceReader.java    | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java
index 8d3de88abb8..bb4d3905ee8 100644
--- a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java
+++ b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java
@@ -83,10 +83,14 @@ record -> {
                                     Object[] values =
                                             new Object[this.rowTypeInfo.getFieldNames().length];
                                     for (int i = 0; i < record.size(); i++) {
-                                        values[i] =
-                                                TypeConvertUtil.valueUnwrap(
-                                                        this.rowTypeInfo.getFieldType(i),
-                                                        record.getValue(i));
+                                        if (record.getValue(i).isNullOrEmpty()) {
+                                            values[i] = null;
+                                        } else {
+                                            values[i] =
+                                                    TypeConvertUtil.valueUnwrap(
+                                                            this.rowTypeInfo.getFieldType(i),
+                                                            record.getValue(i));
+                                        }
                                     }
                                     output.collect(new SeaTunnelRow(values));
                                 });