[Improve][Jdbc] Quotes the identifier for table path (#6951)

Author: hailin0

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/dm/DmdbDialect.java

@@ -131,7 +131,7 @@ public TablePath parse(String tablePath) {
 
     @Override
     public String tableIdentifier(TablePath tablePath) {
-        return tablePath.getSchemaAndTableName();
+        return tablePath.getSchemaAndTableName("\"");
     }
 
     // Compatibility Both database = mode and table-names = schema.tableName are configured

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/iris/IrisDialect.java

@@ -102,7 +102,7 @@ public TablePath parse(String tablePath) {
 
     @Override
     public String tableIdentifier(TablePath tablePath) {
-        return tablePath.getSchemaAndTableName();
+        return quoteIdentifier(tablePath.getSchemaAndTableName());
     }
 
     @Override
@@ -172,7 +172,7 @@ public Object queryNextChunkMax(
                             quotedColumn,
                             chunkSize,
                             quotedColumn,
-                            table.getTablePath().getSchemaAndTableName(),
+                            tableIdentifier(table.getTablePath()),
                             quotedColumn,
                             quotedColumn);
         }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java

@@ -173,7 +173,7 @@ public TablePath parse(String tablePath) {
 
     @Override
     public String tableIdentifier(TablePath tablePath) {
-        return tablePath.getSchemaAndTableName();
+        return quoteIdentifier(tablePath.getSchemaAndTableName());
     }
 
     @Override
@@ -199,7 +199,7 @@ public Long approximateRowCntStatement(Connection connection, JdbcSourceTable ta
             String analyzeTable =
                     String.format(
                             "analyze table %s compute statistics for table",
-                            tablePath.getSchemaAndTableName());
+                            tableIdentifier(tablePath));
             String rowCountQuery =
                     String.format(
                             "select NUM_ROWS from all_tables where OWNER = '%s' AND TABLE_NAME = '%s' ",
@@ -253,7 +253,7 @@ public Object queryNextChunkMax(
                                     + ") WHERE ROWNUM <= %s",
                             quotedColumn,
                             quotedColumn,
-                            table.getTablePath().getSchemaAndTableName(),
+                            tableIdentifier(table.getTablePath()),
                             quotedColumn,
                             quotedColumn,
                             chunkSize);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialect.java

@@ -119,8 +119,8 @@ public Optional<String> getUpsertStatement(
                                 + " UPDATE SET %s"
                                 + " WHEN NOT MATCHED THEN"
                                 + " INSERT (%s) VALUES (%s);",
-                        database,
-                        tableName,
+                        quoteDatabaseIdentifier(database),
+                        quoteIdentifier(tableName),
                         usingClause,
                         onConditions,
                         updateSetClause,
@@ -152,6 +152,11 @@ public String quoteDatabaseIdentifier(String identifier) {
         return "[" + identifier + "]";
     }
 
+    @Override
+    public String tableIdentifier(TablePath tablePath) {
+        return quoteIdentifier(tablePath.getFullName());
+    }
+
     @Override
     public TablePath parse(String tablePath) {
         return TablePath.of(tablePath, true);
@@ -237,7 +242,7 @@ public Object queryNextChunkMax(
                             quotedColumn,
                             chunkSize,
                             quotedColumn,
-                            table.getTablePath().getFullName(),
+                            tableIdentifier(table.getTablePath()),
                             quotedColumn,
                             quotedColumn);
         }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/vertica/VerticaDialect.java

@@ -97,8 +97,8 @@ public Optional<String> getUpsertStatement(
                                 + " UPDATE SET %s"
                                 + " WHEN NOT MATCHED THEN"
                                 + " INSERT (%s) VALUES (%s)",
-                        database,
-                        tableName,
+                        quoteDatabaseIdentifier(database),
+                        quoteIdentifier(tableName),
                         usingClause,
                         onConditions,
                         updateSetClause,

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/xugu/XuguDialect.java

@@ -97,7 +97,7 @@ public TablePath parse(String tablePath) {
 
     @Override
     public String tableIdentifier(TablePath tablePath) {
-        return tablePath.getSchemaAndTableName();
+        return quoteIdentifier(tablePath.getSchemaAndTableName());
     }
 
     @Override