diff --git a/flow/connectors/clickhouse/normalize.go b/flow/connectors/clickhouse/normalize.go
index a77302d0b0..d5375d2902 100644
--- a/flow/connectors/clickhouse/normalize.go
+++ b/flow/connectors/clickhouse/normalize.go
@@ -216,7 +216,8 @@ func (c *ClickHouseConnector) generateCreateTableSQLForNormalizedTable(
 				if err != nil {
 					return nil, fmt.Errorf("error while converting column type to ClickHouse type: %w", err)
 				}
-			} else if (tableSchema.NullableEnabled || columnNullableEnabled) && column.Nullable && !colType.IsArray() {
+			} else if (tableSchema.NullableEnabled || columnNullableEnabled) && column.Nullable && !colType.IsArray() &&
+				!strings.HasPrefix(clickHouseType, "Nullable(") {
 				clickHouseType = fmt.Sprintf("Nullable(%s)", clickHouseType)
 			}
 
diff --git a/flow/connectors/clickhouse/normalize_query.go b/flow/connectors/clickhouse/normalize_query.go
index 83eb14ecbf..44e761dbf6 100644
--- a/flow/connectors/clickhouse/normalize_query.go
+++ b/flow/connectors/clickhouse/normalize_query.go
@@ -168,6 +168,11 @@ func (t *NormalizeQueryGenerator) BuildQuery(ctx context.Context) (string, error
 			if err != nil {
 				return "", fmt.Errorf("error while converting column type to clickhouse type: %w", err)
 			}
+		} else if (schema.NullableEnabled || columnNullableEnabled) && column.Nullable && !colType.IsArray() &&
+			!strings.HasPrefix(clickHouseType, "Nullable(") {
+			// mirror the table DDL: a nullable-enabled column created as Nullable(...) must also be
+			// extracted as Nullable(...), or JSON nulls turn into the type's default value
+			clickHouseType = fmt.Sprintf("Nullable(%s)", clickHouseType)
 		}
 
 		switch clickHouseType {
diff --git a/flow/connectors/clickhouse/normalize_query_test.go b/flow/connectors/clickhouse/normalize_query_test.go
index 71c08c1af4..514317de46 100644
--- a/flow/connectors/clickhouse/normalize_query_test.go
+++ b/flow/connectors/clickhouse/normalize_query_test.go
@@ -9,10 +9,53 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/stretchr/testify/require"
 
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/pkg/testutil"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
+// TestBuildQueryNullableDestinationTypeOverride checks the normalize query agrees with the table DDL on
+// nullability for `destination_type` overrides: with nullability enabled, the DDL creates the column as
+// Nullable(<type>) so the query has to extract Nullable(<type>) too, or JSON nulls turn into the type's
+// default value. An override already spelled Nullable(...) is used as is on both sides.
+func TestBuildQueryNullableDestinationTypeOverride(t *testing.T) {
+	schema := &protos.TableSchema{
+		TableIdentifier:   "src.t1",
+		PrimaryKeyColumns: []string{"id"},
+		System:            protos.TypeSystem_Q,
+		NullableEnabled:   true,
+		Columns: []*protos.FieldDescription{
+			{Name: "id", Type: string(types.QValueKindString), TypeModifier: -1},
+			{Name: "num", Type: string(types.QValueKindInt64), TypeModifier: -1, Nullable: true},
+			{Name: "tag", Type: string(types.QValueKindString), TypeModifier: -1, Nullable: true},
+		},
+	}
+	tableMapping := &protos.TableMapping{
+		SourceTableIdentifier:      "src.t1",
+		DestinationTableIdentifier: "t1_dst",
+		Columns: []*protos.ColumnSetting{
+			{SourceName: "num", DestinationType: "Int64"},
+			{SourceName: "tag", DestinationType: "Nullable(String)"},
+		},
+	}
+
+	query, err := NewNormalizeQueryGenerator(
+		"t1_dst",
+		map[string]*protos.TableSchema{"t1_dst": schema},
+		[]*protos.TableMapping{tableMapping},
+		1, 0,
+		false, false,
+		nil, "_peerdb_raw_t1", nil, false, "", 0, nil,
+	).BuildQuery(t.Context())
+	require.NoError(t, err)
+
+	// nullable-enabled override extracts as Nullable, matching the Nullable(Int64) column the DDL creates
+	require.Contains(t, query, `JSONExtract(_peerdb_data, 'num', 'Nullable(Int64)') AS `+"`num`")
+	// an already-Nullable override is used verbatim, not double wrapped
+	require.Contains(t, query, `JSONExtract(_peerdb_data, 'tag', 'Nullable(String)') AS `+"`tag`")
+	require.NotContains(t, query, "Nullable(Nullable(")
+}
+
 func TestExtendedTimeToDateTime(t *testing.T) {
 	ctx := context.Background()
 	addr := fmt.Sprintf("%s:%d", testutil.ClickHouseTestHost(), testutil.ClickHouseTestPort())