Fix sql federation exception when open db scenario e2e sql test case (#…

…28695)

* Fix sql federation exception when open db scenario e2e sql test case

* Fix wrong view zk path with new metadata structure

* modify maxPoolSize to 5 in db scenario

* modify maxPoolSize to 10 in db scenario

* modify maxPoolSize to 20 in db scenario

* Add log

* Modify maxPoolSize to 50 in db scenario

* Revert "Add log"

This reverts commit 21d9971.

.github/workflows/e2e-sql.yml

@@ -110,7 +110,7 @@ jobs:
         database: [ MySQL, PostgreSQL ]
         # Fix me #25051
         #scenario: [ dbtbl_with_readwrite_splitting, dbtbl_with_readwrite_splitting_and_encrypt, sharding_and_encrypt, encrypt_and_readwrite_splitting, encrypt_shadow, readwrite_splitting_and_shadow, sharding_and_shadow, sharding_encrypt_shadow, mask_encrypt, mask_sharding, mask_encrypt_sharding ]
-        scenario: [ empty_rules, rdl_empty_rules, passthrough, tbl, encrypt, readwrite_splitting, shadow, mask, dbtbl_with_readwrite_splitting_and_encrypt, sharding_and_encrypt, encrypt_and_readwrite_splitting, encrypt_shadow, readwrite_splitting_and_shadow, sharding_and_shadow, sharding_encrypt_shadow, mask_encrypt, mask_sharding, mask_encrypt_sharding ]
+        scenario: [ empty_rules, rdl_empty_rules, passthrough, db, tbl, encrypt, readwrite_splitting, shadow, mask, dbtbl_with_readwrite_splitting_and_encrypt, sharding_and_encrypt, encrypt_and_readwrite_splitting, encrypt_shadow, readwrite_splitting_and_shadow, sharding_and_shadow, sharding_encrypt_shadow, mask_encrypt, mask_sharding, mask_encrypt_sharding ]
         additional-options: [ '' ]
         include:
           - adapter: proxy

kernel/sql-federation/core/src/main/java/org/apache/shardingsphere/sqlfederation/compiler/metadata/util/SQLFederationDataTypeUtils.java

@@ -25,11 +25,14 @@
 import org.apache.calcite.rel.type.RelDataTypeFactory.Builder;
 import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
 import org.apache.shardingsphere.infra.database.mysql.type.MySQLDatabaseType;
+import org.apache.shardingsphere.infra.database.opengauss.type.OpenGaussDatabaseType;
+import org.apache.shardingsphere.infra.database.postgresql.type.PostgreSQLDatabaseType;
 import org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereColumn;
 import org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereTable;
 
 import java.math.BigInteger;
 import java.sql.Types;
+import java.util.Optional;
 
 /**
  * SQL federation data type utility class.
@@ -60,17 +63,33 @@ private static RelDataType getRelDataType(final DatabaseType protocolType, final
     }
 
     private static Class<?> getSqlTypeClass(final DatabaseType protocolType, final ShardingSphereColumn column) {
+        Optional<Class<?>> typeClazz = Optional.empty();
         if (protocolType instanceof MySQLDatabaseType) {
-            if (Types.TINYINT == column.getDataType() || Types.SMALLINT == column.getDataType()) {
-                return Integer.class;
-            }
-            if (Types.INTEGER == column.getDataType()) {
-                return column.isUnsigned() ? Long.class : Integer.class;
-            }
-            if (Types.BIGINT == column.getDataType()) {
-                return column.isUnsigned() ? BigInteger.class : Long.class;
-            }
+            typeClazz = findMySQLTypeClass(column);
         }
-        return SqlType.valueOf(column.getDataType()).clazz;
+        if (protocolType instanceof PostgreSQLDatabaseType || protocolType instanceof OpenGaussDatabaseType) {
+            typeClazz = findPostgreSQLTypeClass(column);
+        }
+        return typeClazz.orElseGet(() -> SqlType.valueOf(column.getDataType()).clazz);
+    }
+
+    private static Optional<Class<?>> findPostgreSQLTypeClass(final ShardingSphereColumn column) {
+        if (Types.SMALLINT == column.getDataType()) {
+            return Optional.of(Integer.class);
+        }
+        return Optional.empty();
+    }
+
+    private static Optional<Class<?>> findMySQLTypeClass(final ShardingSphereColumn column) {
+        if (Types.TINYINT == column.getDataType() || Types.SMALLINT == column.getDataType()) {
+            return Optional.of(Integer.class);
+        }
+        if (Types.INTEGER == column.getDataType()) {
+            return column.isUnsigned() ? Optional.of(Long.class) : Optional.of(Integer.class);
+        }
+        if (Types.BIGINT == column.getDataType()) {
+            return column.isUnsigned() ? Optional.of(BigInteger.class) : Optional.of(Long.class);
+        }
+        return Optional.empty();
     }
 }

kernel/sql-federation/core/src/main/java/org/apache/shardingsphere/sqlfederation/compiler/operator/physical/EnumerableScan.java

@@ -74,7 +74,7 @@ public RelNode copy(final RelTraitSet traitSet, final List<RelNode> inputs) {
 
     @Override
     public RelWriter explainTerms(final RelWriter relWriter) {
-        return super.explainTerms(relWriter).item("sql", sqlString.getSql().replaceAll("[\r\n]", " "));
+        return super.explainTerms(relWriter).item("sql", sqlString.getSql().replaceAll("[\r\n]", " ")).item("dynamicParameters", sqlString.getDynamicParameters());
     }
 
     @Override

kernel/sql-federation/core/src/main/java/org/apache/shardingsphere/sqlfederation/compiler/planner/util/SQLFederationPlannerUtils.java

@@ -247,7 +247,7 @@ public static SqlToRelConverter createSqlToRelConverter(final CalciteCatalogRead
         ViewExpander expander = needsViewExpand ? new ShardingSphereViewExpander(sqlParserRule, databaseType,
                 createSqlToRelConverter(catalogReader, validator, cluster, sqlParserRule, databaseType, false)) : (rowType, queryString, schemaPath, viewPath) -> null;
         // TODO remove withRemoveSortInSubQuery when calcite can expand view which contains order by correctly
-        Config converterConfig = SqlToRelConverter.config().withTrimUnusedFields(true).withRemoveSortInSubQuery(false);
+        Config converterConfig = SqlToRelConverter.config().withTrimUnusedFields(true).withRemoveSortInSubQuery(false).withExpand(true);
         return new SqlToRelConverter(expander, validator, catalogReader, cluster, StandardConvertletTable.INSTANCE, converterConfig);
     }