[feature](jdbc) support mapping varbinary type in JBDC catalog (#58215)

### What problem does this PR solve?
Problem Summary:
support varbinary type mapping in DB2,MYSQL,Oracle,PostgreSQL,SQLServer
JDBC catalog.
u can control this when create catalog with property
"enable.mapping.varbinary", default value is false.
if it's true, will mapping the binary type to doris varbinary type,
if it's false, will mapping the binary type to doris string type.

Followup #57821

### Release note
support mapping varbinary type in JBDC catalog

Author: zhangstar333

docker/thirdparties/docker-compose/oracle/init/03-create-table.sql

@@ -258,3 +258,8 @@ create table doris_test.extreme_test_multi_block (
   t7 interval day(3) to second(6)
 );
 
+CREATE TABLE doris_test.varbinary_test(
+"id" NUMBER(5,0),
+"NAME" VARCHAR2(20),
+"BLOB_COL" BLOB
+);

docker/thirdparties/docker-compose/oracle/init/04-insert.sql

@@ -155,4 +155,8 @@ SELECT
     t1, t2, t3, t4, t5, t6, t7
 FROM doris_test.extreme_test;
 
+INSERT INTO doris_test.varbinary_test VALUES (1, 'empty', EMPTY_BLOB());
+INSERT INTO doris_test.varbinary_test VALUES (2, 'NULL', NULL);
+INSERT INTO doris_test.varbinary_test VALUES (3, 'normal', HEXTORAW('48656C6C6F20576F726C64'));
+
 commit;

fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/DB2JdbcExecutor.java

@@ -81,6 +81,8 @@ protected Object getColumnValue(int columnIndex, ColumnType type, String[] repla
             case VARCHAR:
             case STRING:
                 return resultSet.getObject(columnIndex + 1, String.class);
+            case VARBINARY:
+                return resultSet.getObject(columnIndex + 1, byte[].class);
             default:
                 throw new IllegalArgumentException("Unsupported column type: " + type.getType());
         }

fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/OracleJdbcExecutor.java

@@ -101,6 +101,8 @@ private Object newGetColumnValue(int columnIndex, ColumnType type, String[] repl
             case VARCHAR:
             case STRING:
                 return resultSet.getObject(columnIndex + 1);
+            case VARBINARY:
+                return resultSet.getObject(columnIndex + 1, byte[].class);
             default:
                 throw new IllegalArgumentException("Unsupported column type: " + type.getType());
         }
@@ -142,6 +144,9 @@ private Object oldGetColumnValue(int columnIndex, ColumnType type, String[] repl
             case STRING:
                 Object stringVal = resultSet.getObject(columnIndex + 1);
                 return resultSet.wasNull() ? null : stringVal;
+            case VARBINARY:
+                byte[] bytesVal = resultSet.getBytes(columnIndex + 1);
+                return resultSet.wasNull() ? null : bytesVal;
             default:
                 throw new IllegalArgumentException("Unsupported column type: " + type.getType());
         }

fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/PostgreSQLJdbcExecutor.java

@@ -87,6 +87,8 @@ protected Object getColumnValue(int columnIndex, ColumnType type, String[] repla
             case VARCHAR:
             case STRING:
                 return resultSet.getObject(columnIndex + 1);
+            case VARBINARY:
+                return resultSet.getBytes(columnIndex + 1);
             case ARRAY:
                 java.sql.Array array = resultSet.getArray(columnIndex + 1);
                 return array == null ? null : convertArrayToList(array.getArray());

fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/SQLServerJdbcExecutor.java

@@ -87,6 +87,8 @@ protected Object getColumnValue(int columnIndex, ColumnType type, String[] repla
             case VARCHAR:
             case STRING:
                 return resultSet.getObject(columnIndex + 1);
+            case VARBINARY:
+                return resultSet.getObject(columnIndex + 1, byte[].class);
             default:
                 throw new IllegalArgumentException("Unsupported column type: " + type.getType());
         }

fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/TrinoJdbcExecutor.java

@@ -98,6 +98,8 @@ protected Object getColumnValue(int columnIndex, ColumnType type, String[] repla
                 }
                 return Arrays.asList(dataArray);
             }
+            case VARBINARY:
+                return resultSet.getObject(columnIndex + 1, byte[].class);
             default:
                 throw new IllegalArgumentException("Unsupported column type: " + type.getType());
         }

fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java

@@ -27,6 +27,7 @@
 import org.apache.doris.common.plugin.CloudPluginDownloader.PluginType;
 import org.apache.doris.common.proc.BaseProcResult;
 import org.apache.doris.common.util.Util;
+import org.apache.doris.datasource.CatalogProperty;
 import org.apache.doris.datasource.ExternalCatalog;
 
 import com.google.common.base.Preconditions;
@@ -134,7 +135,8 @@ public class JdbcResource extends Resource {
             CONNECTION_POOL_MAX_WAIT_TIME,
             CONNECTION_POOL_KEEP_ALIVE,
             TEST_CONNECTION,
-            ExternalCatalog.USE_META_CACHE
+            ExternalCatalog.USE_META_CACHE,
+            CatalogProperty.ENABLE_MAPPING_VARBINARY
     ).build();
 
     // The default value of optional properties
@@ -155,6 +157,7 @@ public class JdbcResource extends Resource {
         OPTIONAL_PROPERTIES_DEFAULT_VALUE.put(TEST_CONNECTION, "true");
         OPTIONAL_PROPERTIES_DEFAULT_VALUE.put(ExternalCatalog.USE_META_CACHE,
                 String.valueOf(ExternalCatalog.DEFAULT_USE_META_CACHE));
+        OPTIONAL_PROPERTIES_DEFAULT_VALUE.put(CatalogProperty.ENABLE_MAPPING_VARBINARY, "false");
     }
 
     // timeout for both connection and read. 10 seconds is long enough.

fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/JdbcExternalCatalog.java

@@ -251,7 +251,8 @@ private JdbcClient createJdbcClient() {
                 .setConnectionPoolMaxSize(getConnectionPoolMaxSize())
                 .setConnectionPoolMaxLifeTime(getConnectionPoolMaxLifeTime())
                 .setConnectionPoolMaxWaitTime(getConnectionPoolMaxWaitTime())
-                .setConnectionPoolKeepAlive(isConnectionPoolKeepAlive());
+                .setConnectionPoolKeepAlive(isConnectionPoolKeepAlive())
+                .setEnableMappingVarbinary(getEnableMappingVarbinary());
 
         return JdbcClient.createJdbcClient(jdbcClientConfig);
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcClient.java

@@ -67,6 +67,7 @@ public abstract class JdbcClient {
     protected boolean isOnlySpecifiedDatabase;
     protected Map<String, Boolean> includeDatabaseMap;
     protected Map<String, Boolean> excludeDatabaseMap;
+    protected boolean enableMappingVarbinary;
 
     public static JdbcClient createJdbcClient(JdbcClientConfig jdbcClientConfig) {
         String dbType = parseDbType(jdbcClientConfig.getJdbcUrl());
@@ -111,6 +112,7 @@ protected JdbcClient(JdbcClientConfig jdbcClientConfig) {
         this.dbType = parseDbType(jdbcUrl);
         initializeClassLoader(jdbcClientConfig);
         initializeDataSource(jdbcClientConfig);
+        this.enableMappingVarbinary = jdbcClientConfig.isEnableMappingVarbinary();
     }
 
     protected void setJdbcDriverSystemProperties() {