From 5cc98992723401ab642fd627de3d4f2c1ab6848c Mon Sep 17 00:00:00 2001
From: Chris Cranford <chris@hibernate.org>
Date: Mon, 24 Jun 2024 14:45:19 -0400
Subject: [PATCH] DBZ-7969 Bump Hibernate dependency to 6.4.8.Final

---
 pom.xml                                         |  2 +-
 .../jdbc/dialect/GeneralDatabaseDialect.java    | 17 ++++++++++++++++-
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/pom.xml b/pom.xml
index 009ddb0c..ae12c29c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -28,7 +28,7 @@
         <version.assembly.plugin>3.5.0</version.assembly.plugin>
 
         <!-- Other artifact versions -->
-        <version.hibernate>6.2.18.Final</version.hibernate>
+        <version.hibernate>6.4.8.Final</version.hibernate>
         <version.c3p0>0.9.5.5</version.c3p0>
         <version.assertjdb>2.0.2</version.assertjdb>
 
diff --git a/src/main/java/io/debezium/connector/jdbc/dialect/GeneralDatabaseDialect.java b/src/main/java/io/debezium/connector/jdbc/dialect/GeneralDatabaseDialect.java
index 66bf9a6c..b77011c1 100644
--- a/src/main/java/io/debezium/connector/jdbc/dialect/GeneralDatabaseDialect.java
+++ b/src/main/java/io/debezium/connector/jdbc/dialect/GeneralDatabaseDialect.java
@@ -10,6 +10,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.sql.Types;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.TemporalAccessor;
@@ -503,7 +504,21 @@ public boolean isNegativeScaleAllowed() {
 
     @Override
     public String getTypeName(int jdbcType) {
-        return ddlTypeRegistry.getTypeName(jdbcType, dialect);
+        // To remain consistent with Debezium 2.x releases, the behavior with how column types were
+        // resolved changed in Hibernate 6.3 to align more closely with JPA. This creates an issue
+        // for us as we were relying on Hibernate for column type resolution, and now column types
+        // are being resolved differently. This code aims to retain the Debezium 2.x resolution
+        // functionality.
+        switch (jdbcType) {
+            case Types.VARCHAR:
+                return getTypeName(Types.LONGVARCHAR);
+            case Types.NVARCHAR:
+                return getTypeName(Types.LONGNVARCHAR);
+            case Types.VARBINARY:
+                return getTypeName(Types.LONGVARBINARY);
+            default:
+                return ddlTypeRegistry.getTypeName(jdbcType, dialect);
+        }
     }
 
     @Override