This is an automated email from the ASF dual-hosted git repository.
terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new c11a89b038a Fix PostgreSQL composite column type OID resolution
(#39241)
c11a89b038a is described below
commit c11a89b038ac984f79c28280bcd52280f6857587
Author: Liang Zhang <[email protected]>
AuthorDate: Mon Jul 27 14:26:16 2026 +0800
Fix PostgreSQL composite column type OID resolution (#39241)
* Fix PostgreSQL composite column type OID resolution
* Fix PostgreSQL composite column type OID resolution
* Fix PostgreSQL composite type OID routing
- resolve composite type OIDs from the actual routed data source
- reject ambiguous composite OIDs across multiple data sources
- document error 30031 and fix column metadata argument order
* Fix PostgreSQL composite type OID routing
- resolve composite type OIDs from the actual routed data source
- reject ambiguous composite OIDs across multiple data sources
- document error 30031 and fix column metadata argument order
* Fix PostgreSQL composite type OID routing
- resolve composite type OIDs from the actual routed data source
- reject ambiguous composite OIDs across multiple data sources
- document error 30031 and fix column metadata argument order
* Fix PostgreSQL composite type OID routing
- resolve composite type OIDs from the actual routed data source
- reject ambiguous composite OIDs across multiple data sources
- document error 30031 and fix column metadata argument order
* Fix PostgreSQL composite type OID routing
- resolve composite type OIDs from the actual routed data source
- reject ambiguous composite OIDs across multiple data sources
- document error 30031 and fix column metadata argument order
* Fix PostgreSQL composite type OID routing
- resolve composite type OIDs from the actual routed data source
- reject ambiguous composite OIDs across multiple data sources
- document error 30031 and fix column metadata argument order
* Fix incorrect PostgreSQL composite column type OIDs
- load composite type OIDs from the executed JDBC result connection
- propagate PostgreSQL type OIDs through query header attributes
- reject cross-storage composite results before execution
* Fix incorrect PostgreSQL composite column type OIDs
- load composite type OIDs from the executed JDBC result connection
- propagate PostgreSQL type OIDs through query header attributes
- reject cross-storage composite results before execution
* Fix incorrect PostgreSQL composite column type OIDs
- load composite type OIDs from the executed JDBC result connection
- propagate PostgreSQL type OIDs through query header attributes
- reject cross-storage composite results before execution
---
RELEASE-NOTES.md | 1 +
.../command/query/PostgreSQLColumnDescription.java | 6 +-
.../type/PostgreSQLColumnTypeOIDLoader.java | 81 ++++++++++++++++
.../query/PostgreSQLColumnDescriptionTest.java | 7 ++
.../type/PostgreSQLColumnTypeOIDLoaderTest.java | 84 ++++++++++++++++
.../proxy/backend/connector/ProxySQLExecutor.java | 9 ++
.../connector/StandardDatabaseProxyConnector.java | 12 +--
.../DialectJDBCResultMetadataChecker.java} | 24 +++--
.../backend/response/header/query/QueryHeader.java | 10 ++
.../response/header/query/QueryHeaderBuilder.java | 18 ++++
.../header/query/QueryHeaderBuilderEngine.java | 25 ++++-
.../backend/connector/ProxySQLExecutorTest.java | 34 +++++++
.../StandardDatabaseProxyConnectorTest.java | 10 +-
.../header/query/QueryHeaderBuilderEngineTest.java | 28 +++++-
.../jdbc/PostgreSQLJDBCResultMetadataChecker.java | 66 +++++++++++++
.../header/query/PostgreSQLQueryHeaderBuilder.java | 30 +++++-
....jdbc.executor.DialectJDBCResultMetadataChecker | 18 ++++
.../PostgreSQLJDBCResultMetadataCheckerTest.java | 106 +++++++++++++++++++++
.../query/PostgreSQLQueryHeaderBuilderTest.java | 37 +++++++
.../prepare/MySQLProjectionMetadataResolver.java | 4 +-
.../postgresql/command/query/extended/Portal.java | 7 +-
...PostgreSQLPreparedStatementMetadataFactory.java | 21 +++-
.../describe/PostgreSQLComDescribeExecutor.java | 46 +++++++--
.../query/simple/PostgreSQLComQueryExecutor.java | 8 +-
.../command/query/extended/PortalTest.java | 14 ++-
...greSQLPreparedStatementMetadataFactoryTest.java | 36 +++++++
.../PostgreSQLComDescribeExecutorTest.java | 48 ++++++++--
.../simple/PostgreSQLComQueryExecutorTest.java | 8 +-
28 files changed, 738 insertions(+), 60 deletions(-)
diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index adfc1080407..4a78208c24d 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -36,6 +36,7 @@
1. Proxy: Add MySQL exception mapping for ColumnNotFoundException -
[#39126](https://github.com/apache/shardingsphere/pull/39126)
1. Proxy: Fix MySQL prepared statement parameter signedness decoding -
[#39204](https://github.com/apache/shardingsphere/pull/39204)
1. Proxy: Fix Proxy Native Docker image failing to start due to unexpanded
LOCAL_PATH in ENTRYPOINT -
[#39146](https://github.com/apache/shardingsphere/pull/39146)
+1. Proxy: Fix incorrect PostgreSQL composite column type OIDs in simple and
extended query row descriptions -
[#39241](https://github.com/apache/shardingsphere/pull/39241)
1. JDBC & Proxy: Remove default MySQL prepared statement query properties when
creating data sources -
[#38593](https://github.com/apache/shardingsphere/pull/38593)
1. Mode: Fix rule metadata not removed from memory after dropping rules in
Etcd cluster mode -
[#38561](https://github.com/apache/shardingsphere/pull/38561)
1. Agent: Fix wrong target class name in StaticMethodAdviceExecutor error logs
- [#39077](https://github.com/apache/shardingsphere/pull/39077)
diff --git
a/database/protocol/dialect/postgresql/src/main/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescription.java
b/database/protocol/dialect/postgresql/src/main/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescription.java
index 2c6d16729a0..43ed351fc43 100644
---
a/database/protocol/dialect/postgresql/src/main/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescription.java
+++
b/database/protocol/dialect/postgresql/src/main/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescription.java
@@ -18,6 +18,7 @@
package
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query;
import lombok.Getter;
+import lombok.RequiredArgsConstructor;
import
org.apache.shardingsphere.database.protocol.postgresql.constant.PostgreSQLArrayColumnType;
import
org.apache.shardingsphere.database.protocol.postgresql.constant.PostgreSQLValueFormat;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.extended.PostgreSQLBinaryColumnType;
@@ -27,6 +28,7 @@ import java.sql.Types;
/**
* Column description for PostgreSQL.
*/
+@RequiredArgsConstructor
@Getter
public final class PostgreSQLColumnDescription {
@@ -36,10 +38,10 @@ public final class PostgreSQLColumnDescription {
private final int columnIndex;
- private final int columnLength;
-
private final int typeOID;
+ private final int columnLength;
+
private final int typeModifier = -1;
private final int dataFormat;
diff --git
a/database/protocol/dialect/postgresql/src/main/java/org/apache/shardingsphere/database/protocol/postgresql/type/PostgreSQLColumnTypeOIDLoader.java
b/database/protocol/dialect/postgresql/src/main/java/org/apache/shardingsphere/database/protocol/postgresql/type/PostgreSQLColumnTypeOIDLoader.java
new file mode 100644
index 00000000000..07f1a6ba7e6
--- /dev/null
+++
b/database/protocol/dialect/postgresql/src/main/java/org/apache/shardingsphere/database/protocol/postgresql/type/PostgreSQLColumnTypeOIDLoader.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.database.protocol.postgresql.type;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.postgresql.core.BaseConnection;
+import org.postgresql.core.Oid;
+
+import java.sql.Connection;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Loader for PostgreSQL column type OIDs.
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class PostgreSQLColumnTypeOIDLoader {
+
+ /**
+ * Load composite column type OIDs from result set metadata.
+ *
+ * @param connection database connection
+ * @param metaData result set metadata
+ * @return column indexes to type OIDs, or an empty map if no composite
column type can be resolved
+ * @throws SQLException SQL exception
+ */
+ public static Map<Integer, Integer> load(final Connection connection,
final ResultSetMetaData metaData) throws SQLException {
+ return connection.isWrapperFor(BaseConnection.class) ?
load(connection.unwrap(BaseConnection.class), metaData) :
Collections.emptyMap();
+ }
+
+ private static Map<Integer, Integer> load(final BaseConnection connection,
final ResultSetMetaData metaData) throws SQLException {
+ int columnCount = metaData.getColumnCount();
+ Map<Integer, Integer> result = new HashMap<>();
+ for (int columnIndex = 1; columnIndex <= columnCount; columnIndex++) {
+ if (Types.STRUCT == metaData.getColumnType(columnIndex)) {
+ int typeOID =
connection.getTypeInfo().getPGType(metaData.getColumnTypeName(columnIndex));
+ if (Oid.UNSPECIFIED != typeOID) {
+ result.put(columnIndex, typeOID);
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Find type OID.
+ *
+ * @param connection database connection
+ * @param columnTypeName column type name
+ * @return type OID
+ * @throws SQLException SQL exception
+ */
+ public static Optional<Integer> findTypeOID(final Connection connection,
final String columnTypeName) throws SQLException {
+ if (!connection.isWrapperFor(BaseConnection.class)) {
+ return Optional.empty();
+ }
+ int typeOID =
connection.unwrap(BaseConnection.class).getTypeInfo().getPGType(columnTypeName);
+ return Oid.UNSPECIFIED == typeOID ? Optional.empty() :
Optional.of(typeOID);
+ }
+}
diff --git
a/database/protocol/dialect/postgresql/src/test/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescriptionTest.java
b/database/protocol/dialect/postgresql/src/test/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescriptionTest.java
index a819e62d283..c986e525063 100644
---
a/database/protocol/dialect/postgresql/src/test/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescriptionTest.java
+++
b/database/protocol/dialect/postgresql/src/test/java/org/apache/shardingsphere/database/protocol/postgresql/packet/command/query/PostgreSQLColumnDescriptionTest.java
@@ -26,6 +26,13 @@ import static org.hamcrest.MatcherAssert.assertThat;
class PostgreSQLColumnDescriptionTest {
+ @Test
+ void assertNativeTypeOid() {
+ PostgreSQLColumnDescription description = new
PostgreSQLColumnDescription("record_value", 1, 2249, 32, 0);
+ assertThat(description.getTypeOID(), is(2249));
+ assertThat(description.getColumnLength(), is(32));
+ }
+
@Test
void assertIntegerTypeOid() {
PostgreSQLColumnDescription description = new
PostgreSQLColumnDescription("age", 1, Types.INTEGER, 4, null);
diff --git
a/database/protocol/dialect/postgresql/src/test/java/org/apache/shardingsphere/database/protocol/postgresql/type/PostgreSQLColumnTypeOIDLoaderTest.java
b/database/protocol/dialect/postgresql/src/test/java/org/apache/shardingsphere/database/protocol/postgresql/type/PostgreSQLColumnTypeOIDLoaderTest.java
new file mode 100644
index 00000000000..9d0cfb6b2e4
--- /dev/null
+++
b/database/protocol/dialect/postgresql/src/test/java/org/apache/shardingsphere/database/protocol/postgresql/type/PostgreSQLColumnTypeOIDLoaderTest.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.database.protocol.postgresql.type;
+
+import org.junit.jupiter.api.Test;
+import org.postgresql.core.BaseConnection;
+import org.postgresql.core.Oid;
+import org.postgresql.core.TypeInfo;
+
+import java.sql.Connection;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PostgreSQLColumnTypeOIDLoaderTest {
+
+ @Test
+ void assertLoadFromResultSetMetaData() throws SQLException {
+ BaseConnection connection = mockConnection("record_type", 2249);
+ ResultSetMetaData metaData = mock(ResultSetMetaData.class);
+ when(metaData.getColumnCount()).thenReturn(2);
+ when(metaData.getColumnType(1)).thenReturn(Types.STRUCT);
+ when(metaData.getColumnType(2)).thenReturn(Types.VARCHAR);
+ when(metaData.getColumnTypeName(1)).thenReturn("record_type");
+ Map<Integer, Integer> actual =
PostgreSQLColumnTypeOIDLoader.load(connection, metaData);
+ assertThat(actual, is(Collections.singletonMap(1, 2249)));
+ verify(metaData, never()).getColumnTypeName(2);
+ }
+
+ @Test
+ void assertLoadFromNonPostgreSQLConnection() throws SQLException {
+ Connection connection = mock(Connection.class);
+ assertTrue(PostgreSQLColumnTypeOIDLoader.load(connection,
mock(ResultSetMetaData.class)).isEmpty());
+ }
+
+ @Test
+ void assertFindUnspecifiedTypeOID() throws SQLException {
+ BaseConnection connection = mockConnection("unknown_type",
Oid.UNSPECIFIED);
+ assertFalse(PostgreSQLColumnTypeOIDLoader.findTypeOID(connection,
"unknown_type").isPresent());
+ }
+
+ @Test
+ void assertFindTypeOID() throws SQLException {
+ BaseConnection connection = mockConnection("record_type", 2249);
+ assertThat(PostgreSQLColumnTypeOIDLoader.findTypeOID(connection,
"record_type"), is(Optional.of(2249)));
+ }
+
+ private BaseConnection mockConnection(final String columnTypeName, final
int typeOID) throws SQLException {
+ TypeInfo typeInfo = mock(TypeInfo.class);
+ when(typeInfo.getPGType(columnTypeName)).thenReturn(typeOID);
+ BaseConnection result = mock(BaseConnection.class);
+ when(result.getTypeInfo()).thenReturn(typeInfo);
+ when(result.isWrapperFor(BaseConnection.class)).thenReturn(true);
+ when(result.unwrap(BaseConnection.class)).thenReturn(result);
+ return result;
+ }
+}
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutor.java
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutor.java
index 4e9a3b9c4e2..fe389841780 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutor.java
+++
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutor.java
@@ -56,6 +56,7 @@ import
org.apache.shardingsphere.infra.session.connection.transaction.Transactio
import org.apache.shardingsphere.infra.spi.type.ordered.OrderedSPILoader;
import org.apache.shardingsphere.mode.metadata.MetaDataContexts;
import
org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.ProxyJDBCExecutor;
+import
org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker;
import
org.apache.shardingsphere.proxy.backend.connector.jdbc.statement.JDBCBackendStatement;
import
org.apache.shardingsphere.proxy.backend.connector.sane.DialectSaneQueryResultEngine;
import org.apache.shardingsphere.proxy.backend.context.BackendExecutorContext;
@@ -238,6 +239,14 @@ public final class ProxySQLExecutor {
} catch (final SQLException ex) {
return getSaneExecuteResults(executionContext, ex);
}
+ if (executionContext.getExecutionUnits().size() > 1) {
+ DatabaseType databaseType =
ProxyContext.getInstance().getContextManager().getDatabase(databaseName).getProtocolType();
+ Optional<DialectJDBCResultMetadataChecker> checker =
DatabaseTypedSPILoader.findService(DialectJDBCResultMetadataChecker.class,
databaseType);
+ if (checker.isPresent()) {
+ JDBCExecutionUnit sample =
executionGroupContext.getInputGroups().iterator().next().getInputs().get(0);
+ checker.get().check(executionContext.getExecutionUnits(),
sample.getStorageResource(), sample.getExecutionUnit().getSqlUnit().getSql());
+ }
+ }
executeTransactionHooksBeforeExecuteSQL(databaseConnectionManager.getConnectionSession());
return regularExecutor.execute(executionContext.getQueryContext(),
executionGroupContext, isReturnGeneratedKeys, isExceptionThrown);
}
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnector.java
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnector.java
index 812d30fb76b..b40aeaae72e 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnector.java
+++
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnector.java
@@ -301,10 +301,12 @@ public final class StandardDatabaseProxyConnector
implements DatabaseProxyConnec
int columnCount = getColumnCount(sqlStatementContext,
queryResultSample);
List<QueryHeader> result = new ArrayList<>(columnCount);
QueryHeaderBuilderEngine queryHeaderBuilderEngine = new
QueryHeaderBuilderEngine(database.getProtocolType());
- Collection<ShardingSphereDatabase> databases =
queryContext.getMetaData().getAllDatabases();
ShardingSphereResultSetMetaData resultSetMetaData = new
ShardingSphereResultSetMetaData(queryResultSample.getMetaData().getResultSetMetaData(),
database, sqlStatementContext);
+ Optional<ResultSet> resultSetSample =
cachedResultSets.stream().findFirst();
for (int columnIndex = 1; columnIndex <= columnCount; columnIndex++) {
- result.add(createQueryHeader(queryHeaderBuilderEngine,
sqlStatementContext, resultSetMetaData, database, databases, columnIndex));
+ result.add(resultSetSample.isPresent()
+ ? queryHeaderBuilderEngine.build(sqlStatementContext,
resultSetMetaData, resultSetSample.get(), database, columnIndex)
+ : queryHeaderBuilderEngine.build(sqlStatementContext,
resultSetMetaData, database, columnIndex));
}
return result;
}
@@ -315,12 +317,6 @@ public final class StandardDatabaseProxyConnector
implements DatabaseProxyConnec
: queryResultSample.getMetaData().getColumnCount();
}
- private QueryHeader createQueryHeader(final QueryHeaderBuilderEngine
queryHeaderBuilderEngine, final SQLStatementContext sqlStatementContext,
- final
ShardingSphereResultSetMetaData resultSetMetaData, final ShardingSphereDatabase
database,
- final
Collection<ShardingSphereDatabase> databases, final int columnIndex) throws
SQLException {
- return queryHeaderBuilderEngine.build(sqlStatementContext,
resultSetMetaData, database, databases, columnIndex);
- }
-
private MergedResult mergeQuery(final List<QueryResult> queryResults)
throws SQLException {
MergeEngine mergeEngine = new
MergeEngine(contextManager.getMetaDataContexts().getMetaData(),
database,
contextManager.getMetaDataContexts().getMetaData().getProps(),
databaseConnectionManager.getConnectionSession().getConnectionContext());
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilder.java
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/jdbc/executor/DialectJDBCResultMetadataChecker.java
similarity index 58%
copy from
proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilder.java
copy to
proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/jdbc/executor/DialectJDBCResultMetadataChecker.java
index 0eeeab6658b..d1fa5935a0c 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilder.java
+++
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/connector/jdbc/executor/DialectJDBCResultMetadataChecker.java
@@ -15,31 +15,29 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.proxy.backend.response.header.query;
+package org.apache.shardingsphere.proxy.backend.connector.jdbc.executor;
import org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPI;
-import
org.apache.shardingsphere.driver.jdbc.core.resultset.ShardingSphereResultSetMetaData;
-import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
+import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collection;
/**
- * Query header builder.
+ * Dialect JDBC result metadata checker.
*/
@SingletonSPI
-public interface QueryHeaderBuilder extends DatabaseTypedSPI {
+public interface DialectJDBCResultMetadataChecker extends DatabaseTypedSPI {
/**
- * Build query header.
+ * Check result metadata before execution.
*
- * @param resultSetMetaData result set meta data
- * @param database database
- * @param columnName column name
- * @param columnLabel column label
- * @param columnIndex column index
- * @return query header
+ * @param executionUnits execution units
+ * @param statement JDBC statement
+ * @param sql routed SQL corresponding to the statement
* @throws SQLException SQL exception
*/
- QueryHeader build(ShardingSphereResultSetMetaData resultSetMetaData,
ShardingSphereDatabase database, String columnName, String columnLabel, int
columnIndex) throws SQLException;
+ void check(Collection<ExecutionUnit> executionUnits, Statement statement,
String sql) throws SQLException;
}
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeader.java
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeader.java
index 9feed9ac35e..d8d756b5ef3 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeader.java
+++
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeader.java
@@ -20,6 +20,9 @@ package
org.apache.shardingsphere.proxy.backend.response.header.query;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
+import java.util.Collections;
+import java.util.Map;
+
/**
* Query header.
*/
@@ -50,4 +53,11 @@ public final class QueryHeader {
private final boolean notNull;
private final boolean autoIncrement;
+
+ private final Map<String, Object> protocolAttributes;
+
+ public QueryHeader(final String schema, final String table, final String
columnLabel, final String columnName, final int columnType, final String
columnTypeName,
+ final int columnLength, final int decimals, final
boolean signed, final boolean primaryKey, final boolean notNull, final boolean
autoIncrement) {
+ this(schema, table, columnLabel, columnName, columnType,
columnTypeName, columnLength, decimals, signed, primaryKey, notNull,
autoIncrement, Collections.emptyMap());
+ }
}
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilder.java
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilder.java
index 0eeeab6658b..26a7a92099b 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilder.java
+++
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilder.java
@@ -22,6 +22,7 @@ import
org.apache.shardingsphere.driver.jdbc.core.resultset.ShardingSphereResult
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
+import java.sql.ResultSet;
import java.sql.SQLException;
/**
@@ -42,4 +43,21 @@ public interface QueryHeaderBuilder extends DatabaseTypedSPI
{
* @throws SQLException SQL exception
*/
QueryHeader build(ShardingSphereResultSetMetaData resultSetMetaData,
ShardingSphereDatabase database, String columnName, String columnLabel, int
columnIndex) throws SQLException;
+
+ /**
+ * Build query header.
+ *
+ * @param resultSetMetaData result set meta data
+ * @param resultSet JDBC result set
+ * @param database database
+ * @param columnName column name
+ * @param columnLabel column label
+ * @param columnIndex column index
+ * @return query header
+ * @throws SQLException SQL exception
+ */
+ default QueryHeader build(final ShardingSphereResultSetMetaData
resultSetMetaData, final ResultSet resultSet, final ShardingSphereDatabase
database,
+ final String columnName, final String
columnLabel, final int columnIndex) throws SQLException {
+ return build(resultSetMetaData, database, columnName, columnLabel,
columnIndex);
+ }
}
diff --git
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngine.java
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngine.java
index dd1ec10889e..24a3546bdcc 100644
---
a/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngine.java
+++
b/proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngine.java
@@ -27,8 +27,8 @@ import
org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
import
org.apache.shardingsphere.infra.exception.kernel.syntax.ColumnIndexOutOfRangeException;
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
+import java.sql.ResultSet;
import java.sql.SQLException;
-import java.util.Collection;
/**
* Query header builder engine.
@@ -62,19 +62,38 @@ public final class QueryHeaderBuilderEngine {
* @param sqlStatementContext SQL statement context
* @param resultSetMetaData result set meta data
* @param database current database
- * @param databases available databases
* @param columnIndex column index
* @return query header
* @throws SQLException SQL exception
*/
public QueryHeader build(final SQLStatementContext sqlStatementContext,
final ShardingSphereResultSetMetaData resultSetMetaData, final
ShardingSphereDatabase database,
- final Collection<ShardingSphereDatabase>
databases, final int columnIndex) throws SQLException {
+ final int columnIndex) throws SQLException {
Projection projection = findProjection(sqlStatementContext,
columnIndex);
return null == projection
? build(resultSetMetaData, database, columnIndex)
: queryHeaderBuilder.build(resultSetMetaData, database,
projection.getColumnName(), projection.getColumnLabel(), columnIndex);
}
+ /**
+ * Build query header builder.
+ *
+ * @param sqlStatementContext SQL statement context
+ * @param resultSetMetaData result set meta data
+ * @param resultSet JDBC result set
+ * @param database current database
+ * @param columnIndex column index
+ * @return query header
+ * @throws SQLException SQL exception
+ */
+ public QueryHeader build(final SQLStatementContext sqlStatementContext,
final ShardingSphereResultSetMetaData resultSetMetaData, final ResultSet
resultSet,
+ final ShardingSphereDatabase database, final int
columnIndex) throws SQLException {
+ Projection projection = findProjection(sqlStatementContext,
columnIndex);
+ if (null == projection) {
+ return queryHeaderBuilder.build(resultSetMetaData, resultSet,
database, resultSetMetaData.getColumnName(columnIndex),
resultSetMetaData.getColumnLabel(columnIndex), columnIndex);
+ }
+ return queryHeaderBuilder.build(resultSetMetaData, resultSet,
database, projection.getColumnName(), projection.getColumnLabel(), columnIndex);
+ }
+
private Projection findProjection(final SQLStatementContext
sqlStatementContext, final int columnIndex) {
if (!(sqlStatementContext instanceof SelectStatementContext) ||
!((SelectStatementContext) sqlStatementContext).containsDerivedProjections()) {
return null;
diff --git
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutorTest.java
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutorTest.java
index 9d9a64e68e7..962309c91fc 100644
---
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutorTest.java
+++
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/ProxySQLExecutorTest.java
@@ -32,9 +32,13 @@ import
org.apache.shardingsphere.infra.binder.context.statement.SQLStatementCont
import
org.apache.shardingsphere.infra.binder.context.statement.type.dml.InsertStatementContext;
import org.apache.shardingsphere.infra.config.props.ConfigurationPropertyKey;
import org.apache.shardingsphere.infra.config.rule.RuleConfiguration;
+import org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroup;
import
org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroupContext;
import
org.apache.shardingsphere.infra.executor.kernel.model.ExecutionGroupReportContext;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
+import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
+import org.apache.shardingsphere.infra.executor.sql.context.SQLUnit;
+import
org.apache.shardingsphere.infra.executor.sql.execute.engine.ConnectionMode;
import
org.apache.shardingsphere.infra.executor.sql.execute.engine.driver.jdbc.JDBCExecutionUnit;
import
org.apache.shardingsphere.infra.executor.sql.execute.engine.raw.RawExecutor;
import
org.apache.shardingsphere.infra.executor.sql.execute.engine.raw.RawSQLExecutionUnit;
@@ -54,6 +58,7 @@ import
org.apache.shardingsphere.infra.session.connection.transaction.Transactio
import org.apache.shardingsphere.infra.session.query.QueryContext;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.mode.manager.ContextManager;
+import
org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker;
import
org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.ProxyJDBCExecutor;
import
org.apache.shardingsphere.proxy.backend.connector.jdbc.statement.JDBCBackendStatement;
import
org.apache.shardingsphere.proxy.backend.connector.sane.DialectSaneQueryResultEngine;
@@ -98,6 +103,7 @@ import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import java.sql.SQLException;
+import java.sql.Statement;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -116,6 +122,7 @@ import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.anyCollection;
import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
@@ -315,6 +322,33 @@ class ProxySQLExecutorTest {
Arguments.of("execute-with-driver-and-no-transaction", false,
createInsertStatement(postgresqlDatabaseType), false, false, false));
}
+ @Test
+ void assertExecuteWithDialectResultMetadataCheckerException() throws
SQLException {
+ when(connectionSession.getUsedDatabaseName()).thenReturn("foo_db");
+
when(database.getRuleMetaData().getRules()).thenReturn(createRules(false));
+ ProxySQLExecutor proxySQLExecutor =
createProxySQLExecutor("foo_schema", true);
+ setExecutorField(proxySQLExecutor, "regularExecutor", regularExecutor);
+ ExecutionContext executionContext =
createExecutionContext("execute-with-dialect-result-metadata-checker", new
SQLStatement(fixtureDatabaseType), false);
+ Statement statement = mock(Statement.class);
+ ExecutionUnit executionUnit = new ExecutionUnit("foo_ds", new
SQLUnit("SELECT 1", Collections.emptyList()));
+ Collection<ExecutionUnit> executionUnits =
Arrays.asList(executionUnit, new ExecutionUnit("foo_ds_1", new SQLUnit("SELECT
1", Collections.emptyList())));
+ when(executionContext.getExecutionUnits()).thenReturn(executionUnits);
+ JDBCExecutionUnit jdbcExecutionUnit = new
JDBCExecutionUnit(executionUnit, ConnectionMode.CONNECTION_STRICTLY, statement);
+ ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext = new
ExecutionGroupContext<>(
+ Collections.singletonList(new
ExecutionGroup<>(Collections.singletonList(jdbcExecutionUnit))),
mock(ExecutionGroupReportContext.class));
+ DialectJDBCResultMetadataChecker checker =
mock(DialectJDBCResultMetadataChecker.class);
+ SQLException expected = new SQLException("expected");
+ doThrow(expected).when(checker).check(executionUnits, statement,
"SELECT 1");
+ try (
+ MockedConstruction<DriverExecutionPrepareEngine> ignored =
mockConstruction(DriverExecutionPrepareEngine.class,
+ (mock, context) -> when(mock.prepare(anyString(),
eq(executionContext), anyCollection(),
any(ExecutionGroupReportContext.class))).thenReturn(executionGroupContext));
+ MockedStatic<DatabaseTypedSPILoader> spiLoader =
mockStatic(DatabaseTypedSPILoader.class, CALLS_REAL_METHODS)) {
+ spiLoader.when(() ->
DatabaseTypedSPILoader.findService(DialectJDBCResultMetadataChecker.class,
fixtureDatabaseType)).thenReturn(Optional.of(checker));
+ assertThat(assertThrows(SQLException.class, () ->
proxySQLExecutor.execute(executionContext)), is(expected));
+ verify(regularExecutor, never()).execute(any(), any(),
anyBoolean(), anyBoolean());
+ }
+ }
+
@ParameterizedTest(name = "{0}")
@MethodSource("executeFallbackScenarios")
void assertExecuteFallback(final String name, final boolean
hasRawExecutionRule, final SQLStatement sqlStatement, final boolean
hasSaneResult) throws SQLException {
diff --git
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnectorTest.java
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnectorTest.java
index fd947f30c34..638ddd37c75 100644
---
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnectorTest.java
+++
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/connector/StandardDatabaseProxyConnectorTest.java
@@ -24,6 +24,7 @@ import
org.apache.shardingsphere.database.connector.core.metadata.database.metad
import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
+import
org.apache.shardingsphere.driver.jdbc.core.resultset.ShardingSphereResultSetMetaData;
import
org.apache.shardingsphere.infra.binder.context.segment.insert.keygen.GeneratedKeyContext;
import
org.apache.shardingsphere.infra.binder.context.segment.select.projection.Projection;
import
org.apache.shardingsphere.infra.binder.context.segment.select.projection.ProjectionsContext;
@@ -71,6 +72,7 @@ import
org.apache.shardingsphere.proxy.backend.connector.jdbc.transaction.ProxyB
import org.apache.shardingsphere.proxy.backend.context.BackendExecutorContext;
import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
import org.apache.shardingsphere.proxy.backend.response.header.ResponseHeader;
+import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeaderBuilder;
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryResponseHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.update.UpdateResponseHeader;
@@ -110,6 +112,7 @@ import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
+import java.sql.Types;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -500,12 +503,16 @@ class StandardDatabaseProxyConnectorTest {
when(resultSetMetaData.getColumnLabel(1)).thenReturn("order_id");
when(queryResult.getMetaData()).thenReturn(queryResultMetaData);
when(proxySQLExecutor.execute(executionContext)).thenReturn(Collections.singletonList(queryResult));
+ engine.add(resultSet);
MergedResult mergedResult = mock(MergedResult.class);
when(mergedResult.next()).thenReturn(true);
when(mergedResult.getValue(1, Object.class)).thenReturn(1);
DialectDatabaseMetaData dialectDatabaseMetaData =
mock(DialectDatabaseMetaData.class);
DialectTransactionOption dialectTransactionOption =
mock(DialectTransactionOption.class);
when(dialectDatabaseMetaData.getTransactionOption()).thenReturn(dialectTransactionOption);
+ QueryHeaderBuilder queryHeaderBuilder = mock(QueryHeaderBuilder.class);
+ QueryHeader queryHeader = new QueryHeader("", "", "order_id",
"order_id", Types.INTEGER, "int4", 4, 0, true, false, false, false);
+
when(queryHeaderBuilder.build(any(ShardingSphereResultSetMetaData.class),
eq(resultSet), any(), eq("order_id"), eq("order_id"),
eq(1))).thenReturn(queryHeader);
try (
MockedConstruction<KernelProcessor> mockedKernelProcessor =
mockConstruction(KernelProcessor.class,
(mock, context) ->
when(mock.generateExecutionContext(any(QueryContext.class),
any(RuleMetaData.class),
any(ConfigurationProperties.class))).thenReturn(executionContext));
@@ -515,7 +522,7 @@ class StandardDatabaseProxyConnectorTest {
mockConstruction(MergeEngine.class, (mock, context) ->
when(mock.merge(anyList(), any(QueryContext.class))).thenReturn(mergedResult));
MockedStatic<DatabaseTypedSPILoader> spiLoader =
mockStatic(DatabaseTypedSPILoader.class);
MockedStatic<ShardingSphereServiceLoader> serviceLoader =
mockStatic(ShardingSphereServiceLoader.class)) {
- spiLoader.when(() ->
DatabaseTypedSPILoader.getService(eq(QueryHeaderBuilder.class),
any(DatabaseType.class))).thenReturn(new QueryHeaderBuilderFixture());
+ spiLoader.when(() ->
DatabaseTypedSPILoader.getService(eq(QueryHeaderBuilder.class),
any(DatabaseType.class))).thenReturn(queryHeaderBuilder);
serviceLoader.when(() ->
ShardingSphereServiceLoader.getServiceInstances(AdvancedProxySQLExecutor.class)).thenReturn(Collections.emptyList());
assertThat(engine.execute(), isA(QueryResponseHeader.class));
assertThat(mockedKernelProcessor.constructed().size(), is(1));
@@ -523,6 +530,7 @@ class StandardDatabaseProxyConnectorTest {
assertThat(mockedMergeEngine.constructed().size(), is(1));
assertTrue(engine.next());
assertNotNull(engine.getRowData());
+
verify(queryHeaderBuilder).build(any(ShardingSphereResultSetMetaData.class),
eq(resultSet), any(), eq("order_id"), eq("order_id"), eq(1));
}
}
diff --git
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngineTest.java
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngineTest.java
index 58b13bbf844..df98fc08b3f 100644
---
a/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngineTest.java
+++
b/proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/response/header/query/QueryHeaderBuilderEngineTest.java
@@ -30,6 +30,7 @@ import
org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Collections;
@@ -77,7 +78,26 @@ class QueryHeaderBuilderEngineTest {
QueryHeaderBuilder queryHeaderBuilder =
mock(QueryHeaderBuilder.class);
when(queryHeaderBuilder.build(resultSetMetaData, database, "c1",
"l1", 1)).thenReturn(expectedQueryHeader);
spiLoader.when(() ->
DatabaseTypedSPILoader.getService(QueryHeaderBuilder.class,
databaseType)).thenReturn(queryHeaderBuilder);
- QueryHeader actualQueryHeader = new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
resultSetMetaData, database, Collections.singleton(database), 1);
+ QueryHeader actualQueryHeader = new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
resultSetMetaData, database, 1);
+ assertThat(actualQueryHeader, is(expectedQueryHeader));
+ }
+ }
+
+ @Test
+ void assertBuildWithJDBCResultSet() throws SQLException {
+ SQLStatementContext sqlStatementContext =
mock(SQLStatementContext.class);
+ ShardingSphereResultSetMetaData resultSetMetaData =
mock(ShardingSphereResultSetMetaData.class);
+ when(resultSetMetaData.getColumnName(1)).thenReturn("col_name");
+ when(resultSetMetaData.getColumnLabel(1)).thenReturn("col_label");
+ ResultSet resultSet = mock(ResultSet.class);
+ ShardingSphereDatabase database = mock(ShardingSphereDatabase.class);
+ QueryHeader expectedQueryHeader = mock(QueryHeader.class);
+ try (MockedStatic<DatabaseTypedSPILoader> spiLoader =
mockStatic(DatabaseTypedSPILoader.class)) {
+ QueryHeaderBuilder queryHeaderBuilder =
mock(QueryHeaderBuilder.class);
+ when(queryHeaderBuilder.build(resultSetMetaData, resultSet,
database, "col_name", "col_label", 1)).thenReturn(expectedQueryHeader);
+ spiLoader.when(() ->
DatabaseTypedSPILoader.getService(QueryHeaderBuilder.class,
databaseType)).thenReturn(queryHeaderBuilder);
+ QueryHeader actualQueryHeader =
+ new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
resultSetMetaData, resultSet, database, 1);
assertThat(actualQueryHeader, is(expectedQueryHeader));
}
}
@@ -95,7 +115,7 @@ class QueryHeaderBuilderEngineTest {
try (MockedStatic<DatabaseTypedSPILoader> spiLoader =
mockStatic(DatabaseTypedSPILoader.class)) {
spiLoader.when(() ->
DatabaseTypedSPILoader.getService(QueryHeaderBuilder.class,
databaseType)).thenReturn(mock(QueryHeaderBuilder.class));
assertThrows(ColumnIndexOutOfRangeException.class,
- () -> new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
mock(ShardingSphereResultSetMetaData.class), mock(),
Collections.singleton(mock()), 2));
+ () -> new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
mock(ShardingSphereResultSetMetaData.class), mock(), 2));
}
}
@@ -111,7 +131,7 @@ class QueryHeaderBuilderEngineTest {
QueryHeaderBuilder queryHeaderBuilder =
mock(QueryHeaderBuilder.class);
when(queryHeaderBuilder.build(resultSetMetaData, database,
"col_name", "col_label", 1)).thenReturn(expectedQueryHeader);
spiLoader.when(() ->
DatabaseTypedSPILoader.getService(QueryHeaderBuilder.class,
databaseType)).thenReturn(queryHeaderBuilder);
- QueryHeader actualQueryHeader = new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
resultSetMetaData, database, Collections.singleton(database), 1);
+ QueryHeader actualQueryHeader = new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
resultSetMetaData, database, 1);
assertThat(actualQueryHeader, is(expectedQueryHeader));
}
}
@@ -129,7 +149,7 @@ class QueryHeaderBuilderEngineTest {
try (MockedStatic<DatabaseTypedSPILoader> spiLoader =
mockStatic(DatabaseTypedSPILoader.class)) {
spiLoader.when(() ->
DatabaseTypedSPILoader.getService(QueryHeaderBuilder.class,
databaseType)).thenReturn(mock(QueryHeaderBuilder.class));
assertThrows(ColumnIndexOutOfRangeException.class,
- () -> new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
mock(ShardingSphereResultSetMetaData.class), mock(),
Collections.singleton(mock()), 2));
+ () -> new
QueryHeaderBuilderEngine(databaseType).build(sqlStatementContext,
mock(ShardingSphereResultSetMetaData.class), mock(), 2));
}
}
}
diff --git
a/proxy/backend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/backend/postgresql/connector/jdbc/PostgreSQLJDBCResultMetadataChecker.java
b/proxy/backend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/backend/postgresql/connector/jdbc/PostgreSQLJDBCResultMetadataChecker.java
new file mode 100644
index 00000000000..e6462259def
--- /dev/null
+++
b/proxy/backend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/backend/postgresql/connector/jdbc/PostgreSQLJDBCResultMetadataChecker.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.proxy.backend.postgresql.connector.jdbc;
+
+import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
+import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
+import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
+import
org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Types;
+import java.util.Collection;
+
+/**
+ * JDBC result metadata checker for PostgreSQL.
+ */
+public final class PostgreSQLJDBCResultMetadataChecker implements
DialectJDBCResultMetadataChecker {
+
+ @Override
+ public void check(final Collection<ExecutionUnit> executionUnits, final
Statement statement, final String sql) throws SQLException {
+ if (executionUnits.size() <= 1) {
+ return;
+ }
+ if (statement instanceof PreparedStatement) {
+ checkCompositeType(((PreparedStatement) statement).getMetaData());
+ return;
+ }
+ try (PreparedStatement preparedStatement =
statement.getConnection().prepareStatement(sql)) {
+ checkCompositeType(preparedStatement.getMetaData());
+ }
+ }
+
+ private void checkCompositeType(final ResultSetMetaData metaData) throws
SQLException {
+ if (null == metaData) {
+ return;
+ }
+ int columnCount = metaData.getColumnCount();
+ for (int columnIndex = 1; columnIndex <= columnCount; columnIndex++) {
+ ShardingSpherePreconditions.checkState(Types.STRUCT !=
metaData.getColumnType(columnIndex),
+ () -> new UnsupportedSQLOperationException("PostgreSQL
composite result columns cannot be returned when routed to multiple execution
units"));
+ }
+ }
+
+ @Override
+ public String getDatabaseType() {
+ return "PostgreSQL";
+ }
+}
diff --git
a/proxy/backend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilder.java
b/proxy/backend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilder.java
index 1f61a07f488..a3199c6039f 100644
---
a/proxy/backend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilder.java
+++
b/proxy/backend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilder.java
@@ -17,18 +17,26 @@
package
org.apache.shardingsphere.proxy.backend.postgresql.response.header.query;
+import
org.apache.shardingsphere.database.protocol.postgresql.type.PostgreSQLColumnTypeOIDLoader;
import
org.apache.shardingsphere.driver.jdbc.core.resultset.ShardingSphereResultSetMetaData;
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeaderBuilder;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
/**
* Query header builder for PostgreSQL.
*/
public final class PostgreSQLQueryHeaderBuilder implements QueryHeaderBuilder {
+ public static final String TYPE_OID = "typeOID";
+
private static final int UNUSED_INT_FIELD = 0;
private static final String UNUSED_STRING_FIELD = "";
@@ -38,11 +46,29 @@ public final class PostgreSQLQueryHeaderBuilder implements
QueryHeaderBuilder {
@Override
public QueryHeader build(final ShardingSphereResultSetMetaData
resultSetMetaData, final ShardingSphereDatabase database, final String
columnName, final String columnLabel,
final int columnIndex) throws SQLException {
+ return createQueryHeader(columnLabel,
+ resultSetMetaData.getColumnType(columnIndex),
resultSetMetaData.getColumnTypeName(columnIndex),
resultSetMetaData.getColumnDisplaySize(columnIndex), Collections.emptyMap());
+ }
+
+ @Override
+ public QueryHeader build(final ShardingSphereResultSetMetaData
resultSetMetaData, final ResultSet resultSet, final ShardingSphereDatabase
database, final String columnName,
+ final String columnLabel, final int columnIndex)
throws SQLException {
int columnType = resultSetMetaData.getColumnType(columnIndex);
String columnTypeName =
resultSetMetaData.getColumnTypeName(columnIndex);
- int columnLength = resultSetMetaData.getColumnDisplaySize(columnIndex);
+ return createQueryHeader(columnLabel, columnType, columnTypeName,
resultSetMetaData.getColumnDisplaySize(columnIndex),
getProtocolAttributes(resultSet, columnType, columnTypeName));
+ }
+
+ private QueryHeader createQueryHeader(final String columnLabel, final int
columnType, final String columnTypeName, final int columnLength, final
Map<String, Object> protocolAttributes) {
return new QueryHeader(UNUSED_STRING_FIELD, UNUSED_STRING_FIELD,
columnLabel, UNUSED_STRING_FIELD, columnType, columnTypeName, columnLength,
- UNUSED_INT_FIELD, UNUSED_BOOLEAN_FIELD, UNUSED_BOOLEAN_FIELD,
UNUSED_BOOLEAN_FIELD, UNUSED_BOOLEAN_FIELD);
+ UNUSED_INT_FIELD, UNUSED_BOOLEAN_FIELD, UNUSED_BOOLEAN_FIELD,
UNUSED_BOOLEAN_FIELD, UNUSED_BOOLEAN_FIELD, protocolAttributes);
+ }
+
+ private Map<String, Object> getProtocolAttributes(final ResultSet
resultSet, final int columnType, final String columnTypeName) throws
SQLException {
+ if (Types.STRUCT != columnType) {
+ return Collections.emptyMap();
+ }
+ Optional<Integer> typeOID =
PostgreSQLColumnTypeOIDLoader.findTypeOID(resultSet.getStatement().getConnection(),
columnTypeName);
+ return typeOID.<Map<String, Object>>map(integer ->
Collections.singletonMap(TYPE_OID, integer)).orElse(Collections.emptyMap());
}
@Override
diff --git
a/proxy/backend/dialect/postgresql/src/main/resources/META-INF/services/org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker
b/proxy/backend/dialect/postgresql/src/main/resources/META-INF/services/org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker
new file mode 100644
index 00000000000..cf750c54ea7
--- /dev/null
+++
b/proxy/backend/dialect/postgresql/src/main/resources/META-INF/services/org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.shardingsphere.proxy.backend.postgresql.connector.jdbc.PostgreSQLJDBCResultMetadataChecker
diff --git
a/proxy/backend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/backend/postgresql/connector/jdbc/PostgreSQLJDBCResultMetadataCheckerTest.java
b/proxy/backend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/backend/postgresql/connector/jdbc/PostgreSQLJDBCResultMetadataCheckerTest.java
new file mode 100644
index 00000000000..a1cf1710ac2
--- /dev/null
+++
b/proxy/backend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/backend/postgresql/connector/jdbc/PostgreSQLJDBCResultMetadataCheckerTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.proxy.backend.postgresql.connector.jdbc;
+
+import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
+import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
+import org.apache.shardingsphere.infra.executor.sql.context.SQLUnit;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Types;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PostgreSQLJDBCResultMetadataCheckerTest {
+
+ private final PostgreSQLJDBCResultMetadataChecker checker = new
PostgreSQLJDBCResultMetadataChecker();
+
+ @Test
+ void assertCheckSingleExecutionUnit() throws SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ checker.check(createExecutionUnits("ds_0"), statement, "SELECT
record_value");
+ verify(statement, never()).getMetaData();
+ }
+
+ @Test
+ void assertCheckMultipleExecutionUnits() throws SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ ResultSetMetaData metaData = createMetaData(Types.INTEGER);
+ when(statement.getMetaData()).thenReturn(metaData);
+ assertDoesNotThrow(() -> checker.check(createExecutionUnits("ds_0",
"ds_1"), statement, "SELECT value"));
+ }
+
+ @Test
+ void assertCheckCompositeTypeAcrossExecutionUnits() throws SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ ResultSetMetaData metaData = createMetaData(Types.STRUCT);
+ when(statement.getMetaData()).thenReturn(metaData);
+ UnsupportedSQLOperationException actual =
+ assertThrows(UnsupportedSQLOperationException.class, () ->
checker.check(createExecutionUnits("ds_0", "ds_0"), statement, "SELECT
record_value"));
+ assertThat(actual.getMessage(), is(
+ "Unsupported SQL operation: PostgreSQL composite result
columns cannot be returned when routed to multiple execution units."));
+ }
+
+ @Test
+ void assertCheckStatement() throws SQLException {
+ PreparedStatement preparedStatement = mock(PreparedStatement.class);
+ ResultSetMetaData metaData = createMetaData(Types.INTEGER);
+ when(preparedStatement.getMetaData()).thenReturn(metaData);
+ Connection connection = mock(Connection.class);
+ when(connection.prepareStatement("SELECT
value")).thenReturn(preparedStatement);
+ Statement statement = mock(Statement.class);
+ when(statement.getConnection()).thenReturn(connection);
+ checker.check(createExecutionUnits("ds_0", "ds_1"), statement, "SELECT
value");
+ verify(preparedStatement).close();
+ }
+
+ @Test
+ void assertCheckWithoutResultMetadata() {
+ assertDoesNotThrow(() -> checker.check(createExecutionUnits("ds_0",
"ds_1"), mock(PreparedStatement.class), "UPDATE foo SET value = 1"));
+ }
+
+ private Collection<ExecutionUnit> createExecutionUnits(final String...
storageUnitNames) {
+ ExecutionUnit[] result = new ExecutionUnit[storageUnitNames.length];
+ for (int i = 0; i < storageUnitNames.length; i++) {
+ result[i] = new ExecutionUnit(storageUnitNames[i], new
SQLUnit("SELECT value", Collections.emptyList()));
+ }
+ return Arrays.asList(result);
+ }
+
+ private ResultSetMetaData createMetaData(final int columnType) throws
SQLException {
+ ResultSetMetaData result = mock(ResultSetMetaData.class);
+ when(result.getColumnCount()).thenReturn(1);
+ when(result.getColumnType(1)).thenReturn(columnType);
+ return result;
+ }
+}
diff --git
a/proxy/backend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilderTest.java
b/proxy/backend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilderTest.java
index a9cb26074f7..f8f82b5eb5d 100644
---
a/proxy/backend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilderTest.java
+++
b/proxy/backend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/backend/postgresql/response/header/query/PostgreSQLQueryHeaderBuilderTest.java
@@ -17,16 +17,25 @@
package
org.apache.shardingsphere.proxy.backend.postgresql.response.header.query;
+import
org.apache.shardingsphere.database.protocol.postgresql.type.PostgreSQLColumnTypeOIDLoader;
import
org.apache.shardingsphere.driver.jdbc.core.resultset.ShardingSphereResultSetMetaData;
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import java.sql.Connection;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import java.sql.Types;
+import java.util.Optional;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
class PostgreSQLQueryHeaderBuilderTest {
@@ -45,4 +54,32 @@ class PostgreSQLQueryHeaderBuilderTest {
assertThat(actual.getColumnTypeName(), is("int"));
assertThat(actual.getColumnLength(), is(11));
}
+
+ @Test
+ void assertBuildPostgreSQLCompositeQueryHeader() throws SQLException {
+ int columnIndex = 1;
+ ShardingSphereResultSetMetaData resultSetMetaData =
mock(ShardingSphereResultSetMetaData.class);
+
when(resultSetMetaData.getColumnType(columnIndex)).thenReturn(Types.STRUCT);
+
when(resultSetMetaData.getColumnTypeName(columnIndex)).thenReturn("record_type");
+ ResultSet resultSet = mock(ResultSet.class);
+ Statement statement = mock(Statement.class);
+ Connection connection = mock(Connection.class);
+ when(resultSet.getStatement()).thenReturn(statement);
+ when(statement.getConnection()).thenReturn(connection);
+ try (MockedStatic<PostgreSQLColumnTypeOIDLoader> loader =
mockStatic(PostgreSQLColumnTypeOIDLoader.class)) {
+ loader.when(() ->
PostgreSQLColumnTypeOIDLoader.findTypeOID(connection,
"record_type")).thenReturn(Optional.of(2249));
+ QueryHeader actual = new
PostgreSQLQueryHeaderBuilder().build(resultSetMetaData, resultSet, null, null,
"record", columnIndex);
+
assertThat(actual.getProtocolAttributes().get(PostgreSQLQueryHeaderBuilder.TYPE_OID),
is(2249));
+ }
+ }
+
+ @Test
+ void assertBuildPostgreSQLQueryHeaderFromResultSet() throws SQLException {
+ ShardingSphereResultSetMetaData resultSetMetaData =
mock(ShardingSphereResultSetMetaData.class);
+ when(resultSetMetaData.getColumnType(1)).thenReturn(Types.INTEGER);
+ ResultSet resultSet = mock(ResultSet.class);
+ QueryHeader actual = new
PostgreSQLQueryHeaderBuilder().build(resultSetMetaData, resultSet, null, null,
"id", 1);
+ assertTrue(actual.getProtocolAttributes().isEmpty());
+ verifyNoInteractions(resultSet);
+ }
}
diff --git
a/proxy/frontend/dialect/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLProjectionMetadataResolver.java
b/proxy/frontend/dialect/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLProjectionMetadataResolver.java
index ebd0a75fc32..edbe661f07c 100644
---
a/proxy/frontend/dialect/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLProjectionMetadataResolver.java
+++
b/proxy/frontend/dialect/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLProjectionMetadataResolver.java
@@ -40,7 +40,6 @@ import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashSet;
/**
@@ -78,8 +77,7 @@ public final class MySQLProjectionMetadataResolver {
QueryHeaderBuilderEngine queryHeaderBuilderEngine = new
QueryHeaderBuilderEngine(database.getProtocolType());
Collection<MySQLPacket> result = new
ArrayList<>(selectStatementContext.getProjectionsContext().getExpandProjections().size());
for (int columnIndex = 1; columnIndex <=
selectStatementContext.getProjectionsContext().getExpandProjections().size();
columnIndex++) {
- QueryHeader queryHeader =
- queryHeaderBuilderEngine.build(selectStatementContext,
resultSetMetaData, database, Collections.singleton(database), columnIndex);
+ QueryHeader queryHeader =
queryHeaderBuilderEngine.build(selectStatementContext, resultSetMetaData,
database, columnIndex);
result.add(createMySQLColumnDefinition41Packet(queryHeader,
characterSet));
}
return result;
diff --git
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/Portal.java
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/Portal.java
index 504c97bb980..e4542e185bc 100644
---
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/Portal.java
+++
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/Portal.java
@@ -49,6 +49,7 @@ import
org.apache.shardingsphere.proxy.backend.response.header.ResponseHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryResponseHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.update.UpdateResponseHeader;
+import
org.apache.shardingsphere.proxy.backend.postgresql.response.header.query.PostgreSQLQueryHeaderBuilder;
import
org.apache.shardingsphere.proxy.frontend.postgresql.command.query.PostgreSQLCommand;
import
org.apache.shardingsphere.sql.parser.statement.core.segment.dal.VariableAssignSegment;
import
org.apache.shardingsphere.sql.parser.statement.core.statement.SQLStatement;
@@ -136,7 +137,11 @@ public final class Portal {
int columnIndex = 0;
for (QueryHeader each : queryResponseHeader.getQueryHeaders()) {
PostgreSQLValueFormat valueFormat =
determineValueFormat(columnIndex);
- result.add(new PostgreSQLColumnDescription(each.getColumnLabel(),
++columnIndex, each.getColumnType(), each.getColumnLength(),
each.getColumnTypeName(), valueFormat.getCode()));
+ int currentColumnIndex = ++columnIndex;
+ Integer typeOID = (Integer)
each.getProtocolAttributes().get(PostgreSQLQueryHeaderBuilder.TYPE_OID);
+ result.add(PostgreSQLValueFormat.TEXT == valueFormat && null !=
typeOID
+ ? new PostgreSQLColumnDescription(each.getColumnLabel(),
currentColumnIndex, typeOID, each.getColumnLength(), valueFormat.getCode())
+ : new PostgreSQLColumnDescription(each.getColumnLabel(),
currentColumnIndex, each.getColumnType(), each.getColumnLength(),
each.getColumnTypeName(), valueFormat.getCode()));
}
return result;
}
diff --git
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactory.java
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactory.java
index c0709b3319b..406f9536813 100644
---
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactory.java
+++
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactory.java
@@ -19,18 +19,22 @@ package
org.apache.shardingsphere.proxy.frontend.postgresql.command.query.extend
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
+import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.infra.binder.context.aware.ParameterAware;
import
org.apache.shardingsphere.infra.binder.context.statement.SQLStatementContext;
import org.apache.shardingsphere.infra.binder.engine.SQLBindEngine;
import org.apache.shardingsphere.infra.connection.kernel.KernelProcessor;
import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
+import
org.apache.shardingsphere.infra.exception.external.sql.ShardingSphereSQLException;
import
org.apache.shardingsphere.infra.exception.kernel.metadata.PreparedStatementMetadataResolutionException;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
import
org.apache.shardingsphere.infra.executor.sql.execute.engine.ConnectionMode;
import org.apache.shardingsphere.infra.metadata.ShardingSphereMetaData;
import org.apache.shardingsphere.infra.session.query.QueryContext;
+import org.apache.shardingsphere.infra.util.close.QuietlyCloser;
import
org.apache.shardingsphere.proxy.backend.connector.ProxyDatabaseConnectionManager;
+import
org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker;
import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
@@ -38,6 +42,7 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
+import java.util.Optional;
/**
* Metadata factory for PostgreSQL prepared statements.
@@ -72,6 +77,20 @@ public final class
PostgreSQLPreparedStatementMetadataFactory {
databaseConnectionManager.getConnections(connectionSession.getUsedDatabaseName(),
executionUnit.getDataSourceName(), 0, 1, ConnectionMode.CONNECTION_STRICTLY);
ShardingSpherePreconditions.checkNotEmpty(connections,
() -> new PreparedStatementMetadataResolutionException("no
backend connection was acquired"));
- return
connections.iterator().next().prepareStatement(executionUnit.getSqlUnit().getSql());
+ String sql = executionUnit.getSqlUnit().getSql();
+ PreparedStatement result =
connections.iterator().next().prepareStatement(sql);
+ try {
+ if (executionContext.getExecutionUnits().size() > 1) {
+ Optional<DialectJDBCResultMetadataChecker> checker =
+
DatabaseTypedSPILoader.findService(DialectJDBCResultMetadataChecker.class,
connectionSession.getProtocolType());
+ if (checker.isPresent()) {
+ checker.get().check(executionContext.getExecutionUnits(),
result, sql);
+ }
+ }
+ } catch (final SQLException | ShardingSphereSQLException ex) {
+ QuietlyCloser.close(result);
+ throw ex;
+ }
+ return result;
}
}
diff --git
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutor.java
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutor.java
index 986c476a9d6..411fc417070 100644
---
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutor.java
+++
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutor.java
@@ -21,10 +21,11 @@ import lombok.RequiredArgsConstructor;
import
org.apache.shardingsphere.database.connector.core.type.DatabaseTypeRegistry;
import
org.apache.shardingsphere.database.exception.core.exception.syntax.column.ColumnNotFoundException;
import org.apache.shardingsphere.database.protocol.packet.DatabasePacket;
-import
org.apache.shardingsphere.database.protocol.postgresql.packet.PostgreSQLPacket;
+import
org.apache.shardingsphere.database.protocol.postgresql.constant.PostgreSQLValueFormat;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLColumnDescription;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLNoDataPacket;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLRowDescriptionPacket;
+import
org.apache.shardingsphere.database.protocol.postgresql.type.PostgreSQLColumnTypeOIDLoader;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.extended.PostgreSQLBinaryColumnType;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.extended.describe.PostgreSQLComDescribePacket;
import org.apache.shardingsphere.infra.exception.ShardingSpherePreconditions;
@@ -60,9 +61,9 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
-import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -111,7 +112,7 @@ public final class PostgreSQLComDescribeExecutor implements
CommandExecutor {
}
}
- private void describeInsertStatementByDatabaseMetaData(final
PostgreSQLServerPreparedStatement preparedStatement) {
+ private void describeInsertStatementByDatabaseMetaData(final
PostgreSQLServerPreparedStatement preparedStatement) throws SQLException {
InsertStatement insertStatement = (InsertStatement)
preparedStatement.getSqlStatementContext().getSqlStatement();
Collection<Integer> unspecifiedTypeParameterIndexes =
getUnspecifiedTypeParameterIndexes(preparedStatement);
Optional<ReturningSegment> returningSegment =
insertStatement.getReturning();
@@ -121,7 +122,9 @@ public final class PostgreSQLComDescribeExecutor implements
CommandExecutor {
String logicTableName = insertStatement.getTable().map(optional ->
optional.getTableName().getIdentifier().getValue()).orElse("");
ShardingSphereTable table =
getTableFromMetaData(connectionSession.getUsedDatabaseName(), insertStatement,
logicTableName);
List<ShardingSphereIdentifier> columnNamesOfInsert =
getColumnNamesOfInsertStatement(insertStatement, table);
-
preparedStatement.setRowDescription(returningSegment.<PostgreSQLPacket>map(returning
-> describeReturning(returning,
table)).orElseGet(PostgreSQLNoDataPacket::getInstance));
+ preparedStatement.setRowDescription(returningSegment.isPresent()
+ ? describeReturning(returningSegment.get(), table,
preparedStatement)
+ : PostgreSQLNoDataPacket.getInstance());
int parameterMarkerIndex = 0;
for (InsertValuesSegment each : insertStatement.getValues()) {
for (int i = 0; i < each.getValues().size(); i++) {
@@ -164,8 +167,9 @@ public final class PostgreSQLComDescribeExecutor implements
CommandExecutor {
: insertStatement.getColumns().stream().map(each -> new
ShardingSphereIdentifier(each.getIdentifier().getValue())).collect(Collectors.toList());
}
- private PostgreSQLRowDescriptionPacket describeReturning(final
ReturningSegment returningSegment, final ShardingSphereTable table) {
- Collection<PostgreSQLColumnDescription> result = new LinkedList<>();
+ private PostgreSQLRowDescriptionPacket describeReturning(final
ReturningSegment returningSegment, final ShardingSphereTable table,
+ final
PostgreSQLServerPreparedStatement preparedStatement) throws SQLException {
+ List<PostgreSQLColumnDescription> result = new ArrayList<>();
for (ProjectionSegment each :
returningSegment.getProjections().getProjections()) {
if (each instanceof ShorthandProjectionSegment) {
table.getAllColumns().stream()
@@ -182,6 +186,7 @@ public final class PostgreSQLComDescribeExecutor implements
CommandExecutor {
result.add(convertExpressionToDescription((ExpressionProjectionSegment) each));
}
}
+ applyColumnTypeOIDs(result, loadColumnTypeOIDs(preparedStatement,
result.size()));
return new PostgreSQLRowDescriptionPacket(result);
}
@@ -223,6 +228,15 @@ public final class PostgreSQLComDescribeExecutor
implements CommandExecutor {
}
}
+ private Map<Integer, Integer> loadColumnTypeOIDs(final
PostgreSQLServerPreparedStatement preparedStatement, final int
expectedColumnCount) throws SQLException {
+ try (PreparedStatement actualPreparedStatement =
PostgreSQLPreparedStatementMetadataFactory.load(connectionSession,
preparedStatement, Collections.emptyList())) {
+ ResultSetMetaData metaData = actualPreparedStatement.getMetaData();
+ return null == metaData || expectedColumnCount !=
metaData.getColumnCount()
+ ? Collections.emptyMap()
+ :
PostgreSQLColumnTypeOIDLoader.load(actualPreparedStatement.getConnection(),
metaData);
+ }
+ }
+
private void tryDescribePreparedStatementByJDBC(final
PostgreSQLServerPreparedStatement logicPreparedStatement) throws SQLException {
try (PreparedStatement actualPreparedStatement =
PostgreSQLPreparedStatementMetadataFactory.load(connectionSession,
logicPreparedStatement, Collections.emptyList())) {
PostgreSQLPreparedStatementParameterTypeResolver.resolveParameterTypes(logicPreparedStatement,
actualPreparedStatement);
@@ -239,14 +253,32 @@ public final class PostgreSQLComDescribeExecutor
implements CommandExecutor {
logicPreparedStatement.setRowDescription(PostgreSQLNoDataPacket.getInstance());
return;
}
+ Map<Integer, Integer> columnTypeOIDs =
PostgreSQLColumnTypeOIDLoader.load(actualPreparedStatement.getConnection(),
resultSetMetaData);
List<PostgreSQLColumnDescription> columnDescriptions = new
ArrayList<>(resultSetMetaData.getColumnCount());
for (int columnIndex = 1; columnIndex <=
resultSetMetaData.getColumnCount(); columnIndex++) {
String columnName = resultSetMetaData.getColumnName(columnIndex);
int columnType = resultSetMetaData.getColumnType(columnIndex);
int columnLength =
resultSetMetaData.getColumnDisplaySize(columnIndex);
String columnTypeName =
resultSetMetaData.getColumnTypeName(columnIndex);
- columnDescriptions.add(new PostgreSQLColumnDescription(columnName,
columnIndex, columnType, columnLength, columnTypeName));
+ Integer typeOID = columnTypeOIDs.get(columnIndex);
+ columnDescriptions.add(null != typeOID
+ ? new PostgreSQLColumnDescription(columnName, columnIndex,
typeOID, columnLength, PostgreSQLValueFormat.TEXT.getCode())
+ : new PostgreSQLColumnDescription(columnName, columnIndex,
columnType, columnLength, columnTypeName));
}
logicPreparedStatement.setRowDescription(new
PostgreSQLRowDescriptionPacket(columnDescriptions));
}
+
+ private void applyColumnTypeOIDs(final List<PostgreSQLColumnDescription>
columnDescriptions, final Map<Integer, Integer> columnTypeOIDs) {
+ if (columnTypeOIDs.isEmpty()) {
+ return;
+ }
+ for (int columnIndex = 1; columnIndex <= columnDescriptions.size();
columnIndex++) {
+ Integer typeOID = columnTypeOIDs.get(columnIndex);
+ if (null != typeOID) {
+ PostgreSQLColumnDescription columnDescription =
columnDescriptions.get(columnIndex - 1);
+ columnDescriptions.set(columnIndex - 1, new
PostgreSQLColumnDescription(
+ columnDescription.getColumnName(),
columnDescription.getColumnIndex(), typeOID,
columnDescription.getColumnLength(), columnDescription.getDataFormat()));
+ }
+ }
+ }
}
diff --git
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutor.java
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutor.java
index 90eb68e68f1..1dfef9fe909 100644
---
a/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutor.java
+++
b/proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutor.java
@@ -21,6 +21,7 @@ import lombok.Getter;
import
org.apache.shardingsphere.database.connector.core.metadata.database.enums.QuoteCharacter;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import org.apache.shardingsphere.database.protocol.packet.DatabasePacket;
+import
org.apache.shardingsphere.database.protocol.postgresql.constant.PostgreSQLValueFormat;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.PostgreSQLPacket;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLColumnDescription;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLDataRowPacket;
@@ -38,6 +39,7 @@ import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryResponseHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.update.UpdateResponseHeader;
import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
+import
org.apache.shardingsphere.proxy.backend.postgresql.response.header.query.PostgreSQLQueryHeaderBuilder;
import
org.apache.shardingsphere.proxy.frontend.command.executor.QueryCommandExecutor;
import org.apache.shardingsphere.proxy.frontend.command.executor.ResponseType;
import
org.apache.shardingsphere.proxy.frontend.postgresql.command.PortalContext;
@@ -98,7 +100,11 @@ public final class PostgreSQLComQueryExecutor implements
QueryCommandExecutor {
Collection<PostgreSQLColumnDescription> result = new LinkedList<>();
int columnIndex = 0;
for (QueryHeader each : queryResponseHeader.getQueryHeaders()) {
- result.add(new PostgreSQLColumnDescription(each.getColumnLabel(),
++columnIndex, each.getColumnType(), each.getColumnLength(),
each.getColumnTypeName()));
+ int currentColumnIndex = ++columnIndex;
+ Integer typeOID = (Integer)
each.getProtocolAttributes().get(PostgreSQLQueryHeaderBuilder.TYPE_OID);
+ result.add(null == typeOID
+ ? new PostgreSQLColumnDescription(each.getColumnLabel(),
currentColumnIndex, each.getColumnType(), each.getColumnLength(),
each.getColumnTypeName())
+ : new PostgreSQLColumnDescription(each.getColumnLabel(),
currentColumnIndex, typeOID, each.getColumnLength(),
PostgreSQLValueFormat.TEXT.getCode()));
}
return result;
}
diff --git
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PortalTest.java
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PortalTest.java
index 1aa91f84c6a..b026d5f3b06 100644
---
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PortalTest.java
+++
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PortalTest.java
@@ -52,6 +52,7 @@ import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryResponseHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.update.UpdateResponseHeader;
import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
+import
org.apache.shardingsphere.proxy.backend.postgresql.response.header.query.PostgreSQLQueryHeaderBuilder;
import
org.apache.shardingsphere.sql.parser.statement.core.segment.dal.VariableAssignSegment;
import
org.apache.shardingsphere.sql.parser.statement.core.segment.dal.VariableSegment;
import
org.apache.shardingsphere.sql.parser.statement.core.statement.SQLStatement;
@@ -136,9 +137,10 @@ class PortalTest {
@Test
void assertExecuteSelectStatementAndReturnAllRows() throws SQLException,
ReflectiveOperationException {
QueryResponseHeader responseHeader = mock(QueryResponseHeader.class);
- QueryHeader queryHeader = new QueryHeader("schema", "table",
"columnLabel", "columnName", Types.VARCHAR, "columnTypeName", 0, 0, false,
false, false, false);
- QueryHeader intColumnQueryHeader = new QueryHeader("schema", "table",
"columnLabel", "columnName", Types.INTEGER, "columnTypeName", 0, 0, false,
false, false, false);
-
when(responseHeader.getQueryHeaders()).thenReturn(Arrays.asList(queryHeader,
intColumnQueryHeader));
+ QueryHeader queryHeader = new QueryHeader("schema", "table",
"columnLabel", "columnName", Types.STRUCT, "record_type", 0, 0, false, false,
false, false,
+
Collections.singletonMap(PostgreSQLQueryHeaderBuilder.TYPE_OID, 2249));
+ QueryHeader binaryQueryHeader = new QueryHeader("schema", "table",
"columnLabel", "columnName", Types.STRUCT, "record_type", 0, 0, false, false,
false, false);
+
when(responseHeader.getQueryHeaders()).thenReturn(Arrays.asList(queryHeader,
binaryQueryHeader));
when(proxyBackendHandler.execute()).thenReturn(responseHeader);
when(proxyBackendHandler.next()).thenReturn(true, true, false);
when(proxyBackendHandler.getRowData()).thenReturn(new
QueryResponseRow(Collections.singletonList(new QueryResponseCell(Types.INTEGER,
0))),
@@ -153,9 +155,11 @@ class PortalTest {
.get(PostgreSQLRowDescriptionPacket.class.getDeclaredField("columnDescriptions"),
portalDescription);
Iterator<PostgreSQLColumnDescription> columnDescriptionIterator =
columnDescriptions.iterator();
PostgreSQLColumnDescription textColumnDescription =
columnDescriptionIterator.next();
- PostgreSQLColumnDescription intColumnDescription =
columnDescriptionIterator.next();
+ PostgreSQLColumnDescription binaryColumnDescription =
columnDescriptionIterator.next();
+ assertThat(textColumnDescription.getTypeOID(), is(2249));
assertThat(textColumnDescription.getDataFormat(),
is(PostgreSQLValueFormat.TEXT.getCode()));
- assertThat(intColumnDescription.getDataFormat(),
is(PostgreSQLValueFormat.BINARY.getCode()));
+ assertThat(binaryColumnDescription.getTypeOID(),
is(PostgreSQLBinaryColumnType.VARCHAR.getValue()));
+ assertThat(binaryColumnDescription.getDataFormat(),
is(PostgreSQLValueFormat.BINARY.getCode()));
List<DatabasePacket> actualPackets = portal.execute(0);
assertThat(actualPackets.size(), is(3));
assertThat(actualPackets.get(0), isA(PostgreSQLDataRowPacket.class));
diff --git
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactoryTest.java
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactoryTest.java
index 6881558cdbc..b2c47e89fd4 100644
---
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactoryTest.java
+++
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/PostgreSQLPreparedStatementMetadataFactoryTest.java
@@ -17,6 +17,7 @@
package
org.apache.shardingsphere.proxy.frontend.postgresql.command.query.extended;
+import
org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.extended.PostgreSQLBinaryColumnType;
import
org.apache.shardingsphere.infra.binder.context.statement.SQLStatementContext;
@@ -24,8 +25,11 @@ import
org.apache.shardingsphere.infra.binder.context.statement.type.dml.SelectS
import org.apache.shardingsphere.infra.config.props.ConfigurationProperties;
import org.apache.shardingsphere.infra.connection.kernel.KernelProcessor;
import
org.apache.shardingsphere.infra.exception.external.sql.ShardingSphereSQLException;
+import
org.apache.shardingsphere.infra.exception.generic.UnsupportedSQLOperationException;
import
org.apache.shardingsphere.infra.exception.kernel.metadata.PreparedStatementMetadataResolutionException;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
+import org.apache.shardingsphere.infra.executor.sql.context.ExecutionUnit;
+import org.apache.shardingsphere.infra.executor.sql.context.SQLUnit;
import org.apache.shardingsphere.infra.hint.HintValueContext;
import
org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
import
org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnit;
@@ -38,6 +42,7 @@ import
org.apache.shardingsphere.infra.session.connection.ConnectionContext;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.mode.manager.ContextManager;
import
org.apache.shardingsphere.proxy.backend.connector.ProxyDatabaseConnectionManager;
+import
org.apache.shardingsphere.proxy.backend.connector.jdbc.executor.DialectJDBCResultMetadataChecker;
import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
import org.apache.shardingsphere.sql.parser.engine.api.CacheOption;
@@ -51,13 +56,16 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.Properties;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -68,9 +76,13 @@ import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(AutoMockExtension.class)
@@ -115,11 +127,35 @@ class PostgreSQLPreparedStatementMetadataFactoryTest {
}
}
+ @Test
+ void assertLoadWithMetadataCheckerException() throws SQLException {
+ PreparedStatement expected = prepareJDBCBackendConnection(null);
+ PostgreSQLServerPreparedStatement preparedStatement =
createPreparedStatement(true);
+ ExecutionContext executionContext = mock(ExecutionContext.class);
+ List<ExecutionUnit> executionUnits = Arrays.asList(
+ new ExecutionUnit("ds_0", new SQLUnit("SELECT id FROM foo_tbl
WHERE id=?", PARAMETERS)),
+ new ExecutionUnit("ds_1", new SQLUnit("SELECT id FROM foo_tbl
WHERE id=?", PARAMETERS)));
+ when(executionContext.getExecutionUnits()).thenReturn(executionUnits);
+ DialectJDBCResultMetadataChecker checker =
mock(DialectJDBCResultMetadataChecker.class);
+ UnsupportedSQLOperationException expectedException = new
UnsupportedSQLOperationException("expected");
+ doThrow(expectedException).when(checker).check(executionUnits,
expected, "SELECT id FROM foo_tbl WHERE id=?");
+ try (
+ MockedConstruction<KernelProcessor> ignored =
mockConstruction(KernelProcessor.class,
+ (mock, context) ->
when(mock.generateExecutionContext(any(), any(),
any())).thenReturn(executionContext));
+ MockedStatic<DatabaseTypedSPILoader> spiLoader =
mockStatic(DatabaseTypedSPILoader.class, CALLS_REAL_METHODS)) {
+ spiLoader.when(() ->
DatabaseTypedSPILoader.findService(DialectJDBCResultMetadataChecker.class,
databaseType)).thenReturn(Optional.of(checker));
+ assertThat(assertThrows(UnsupportedSQLOperationException.class,
+ () ->
PostgreSQLPreparedStatementMetadataFactory.load(connectionSession,
preparedStatement, PARAMETERS)), is(expectedException));
+ verify(expected).close();
+ }
+ }
+
private PostgreSQLServerPreparedStatement createPreparedStatement(final
boolean withUsedDatabaseName) {
SQLStatement sqlStatement = sqlParserEngine.parse("SELECT id FROM
foo_tbl WHERE id=?", false);
SQLStatementContext sqlStatementContext =
mock(SelectStatementContext.class);
when(sqlStatementContext.getSqlStatement()).thenReturn(sqlStatement);
when(connectionSession.getCurrentDatabaseName()).thenReturn("postgres");
+
lenient().when(connectionSession.getProtocolType()).thenReturn(databaseType);
if (withUsedDatabaseName) {
when(connectionSession.getUsedDatabaseName()).thenReturn("postgres");
}
diff --git
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutorTest.java
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutorTest.java
index a97d31e2b02..e08465f1b09 100644
---
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutorTest.java
+++
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/extended/describe/PostgreSQLComDescribeExecutorTest.java
@@ -21,6 +21,7 @@ import lombok.SneakyThrows;
import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
import
org.apache.shardingsphere.database.exception.core.exception.syntax.column.ColumnNotFoundException;
import org.apache.shardingsphere.database.protocol.packet.DatabasePacket;
+import
org.apache.shardingsphere.database.protocol.postgresql.constant.PostgreSQLValueFormat;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLColumnDescription;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLNoDataPacket;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.PostgreSQLParameterDescriptionPacket;
@@ -28,6 +29,7 @@ import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.que
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.extended.PostgreSQLBinaryColumnType;
import
org.apache.shardingsphere.database.protocol.postgresql.packet.command.query.extended.describe.PostgreSQLComDescribePacket;
import
org.apache.shardingsphere.database.protocol.postgresql.payload.PostgreSQLPacketPayload;
+import
org.apache.shardingsphere.database.protocol.postgresql.type.PostgreSQLColumnTypeOIDLoader;
import
org.apache.shardingsphere.infra.binder.context.statement.SQLStatementContext;
import
org.apache.shardingsphere.infra.binder.context.statement.type.dml.InsertStatementContext;
import
org.apache.shardingsphere.infra.binder.context.statement.type.dml.SelectStatementContext;
@@ -60,6 +62,7 @@ import
org.apache.shardingsphere.sqltranslator.rule.SQLTranslatorRule;
import
org.apache.shardingsphere.sqltranslator.rule.builder.DefaultSQLTranslatorRuleConfigurationBuilder;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
import
org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
@@ -81,6 +84,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -106,7 +110,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(AutoMockExtension.class)
-@StaticMockSettings(ProxyContext.class)
+@StaticMockSettings({ProxyContext.class, PostgreSQLColumnTypeOIDLoader.class})
@MockitoSettings(strictness = Strictness.LENIENT)
class PostgreSQLComDescribeExecutorTest {
@@ -127,9 +131,18 @@ class PostgreSQLComDescribeExecutorTest {
@Mock
private ConnectionSession connectionSession;
+ private Map<Integer, Integer> columnTypeOIDs;
+
@InjectMocks
private PostgreSQLComDescribeExecutor executor;
+ @BeforeEach
+ void setUp() throws SQLException {
+ when(connectionSession.getProtocolType()).thenReturn(DATABASE_TYPE);
+ columnTypeOIDs = Collections.emptyMap();
+ when(PostgreSQLColumnTypeOIDLoader.load(any(Connection.class),
any(ResultSetMetaData.class))).thenAnswer(ignored -> columnTypeOIDs);
+ }
+
@Test
void assertDescribePortal() throws SQLException {
when(packet.getType()).thenReturn('P');
@@ -254,7 +267,6 @@ class PostgreSQLComDescribeExecutorTest {
when(packet.getName()).thenReturn(statementId);
String sql = "INSERT INTO public.t_small (col1, col2) VALUES (?, ?)
RETURNING *, col1 + col2 expr_sum";
SQLStatement sqlStatement = SQL_PARSER_ENGINE.parse(sql, false);
- List<PostgreSQLBinaryColumnType> parameterTypes =
Arrays.asList(PostgreSQLBinaryColumnType.INT4,
PostgreSQLBinaryColumnType.UNSPECIFIED);
SQLStatementContext sqlStatementContext =
mock(InsertStatementContext.class);
when(sqlStatementContext.getSqlStatement()).thenReturn(sqlStatement);
ShardingSphereTable table = new ShardingSphereTable("t_small",
@@ -264,6 +276,8 @@ class PostgreSQLComDescribeExecutorTest {
Collections.emptyList(), Collections.emptyList());
ContextManager contextManager = mockContextManager(table);
when(ProxyContext.getInstance().getContextManager()).thenReturn(contextManager);
+ prepareJDBCBackendConnectionForResultMetaData(sql, 3);
+ List<PostgreSQLBinaryColumnType> parameterTypes =
Arrays.asList(PostgreSQLBinaryColumnType.INT4,
PostgreSQLBinaryColumnType.UNSPECIFIED);
List<Integer> parameterIndexes = IntStream.range(0,
sqlStatement.getParameterCount()).boxed().collect(Collectors.toList());
connectionSession.getServerPreparedStatementRegistry().addPreparedStatement(
statementId, new PostgreSQLServerPreparedStatement(sql,
sqlStatementContext, new HintValueContext(), parameterTypes, parameterIndexes));
@@ -377,6 +391,12 @@ class PostgreSQLComDescribeExecutorTest {
when(sqlStatementContext.getSqlStatement()).thenReturn(sqlStatement);
ContextManager contextManager = mockContextManager();
when(ProxyContext.getInstance().getContextManager()).thenReturn(contextManager);
+ prepareJDBCBackendConnectionForResultMetaData(sql,
expectedColumns.size());
+ if ("returning complex columns".equals(testName)) {
+ columnTypeOIDs = new HashMap<>(2, 1F);
+ columnTypeOIDs.put(12, 2249);
+ columnTypeOIDs.put(13, 2249);
+ }
List<Integer> parameterIndexes = IntStream.range(0,
sqlStatement.getParameterCount()).boxed().collect(Collectors.toList());
connectionSession.getServerPreparedStatementRegistry().addPreparedStatement(
statementId, new PostgreSQLServerPreparedStatement(sql,
sqlStatementContext, new HintValueContext(), parameterTypes, parameterIndexes));
@@ -413,13 +433,14 @@ class PostgreSQLComDescribeExecutorTest {
SQLStatementContext sqlStatementContext =
mock(SelectStatementContext.class);
when(sqlStatementContext.getSqlStatement()).thenReturn(sqlStatement);
prepareJDBCBackendConnection(sql);
- List<PostgreSQLBinaryColumnType> parameterTypes = new
ArrayList<>(Collections.singleton(PostgreSQLBinaryColumnType.UNSPECIFIED));
ContextManager contextManager = mockContextManager();
when(ProxyContext.getInstance().getContextManager()).thenReturn(contextManager);
- List<Integer> parameterIndexes = IntStream.range(0,
sqlStatement.getParameterCount()).boxed().collect(Collectors.toList());
ConnectionContext connectionContext = mock(ConnectionContext.class);
when(connectionContext.getCurrentDatabaseName()).thenReturn(Optional.of(DATABASE_NAME));
when(connectionSession.getConnectionContext()).thenReturn(connectionContext);
+ columnTypeOIDs = Collections.singletonMap(1, 2249);
+ List<PostgreSQLBinaryColumnType> parameterTypes = new
ArrayList<>(Collections.singleton(PostgreSQLBinaryColumnType.UNSPECIFIED));
+ List<Integer> parameterIndexes = IntStream.range(0,
sqlStatement.getParameterCount()).boxed().collect(Collectors.toList());
connectionSession.getServerPreparedStatementRegistry().addPreparedStatement(
statementId, new PostgreSQLServerPreparedStatement(sql,
sqlStatementContext, new HintValueContext(), parameterTypes, parameterIndexes));
Collection<DatabasePacket> actual = executor.execute();
@@ -434,7 +455,7 @@ class PostgreSQLComDescribeExecutorTest {
PostgreSQLRowDescriptionPacket actualRowDescription =
(PostgreSQLRowDescriptionPacket) actualPacketsIterator.next();
List<PostgreSQLColumnDescription> actualColumnDescriptions =
getColumnDescriptions(actualRowDescription);
List<PostgreSQLColumnDescription> expectedColumnDescriptions =
Arrays.asList(
- new PostgreSQLColumnDescription("id", 1, Types.INTEGER, 11,
"int4"),
+ new PostgreSQLColumnDescription("id", 1, 2249, 11,
PostgreSQLValueFormat.TEXT.getCode()),
new PostgreSQLColumnDescription("k", 2, Types.INTEGER, 11,
"int4"),
new PostgreSQLColumnDescription("c", 3, Types.CHAR, 60,
"int4"),
new PostgreSQLColumnDescription("pad", 4, Types.CHAR, 120,
"int4"));
@@ -644,6 +665,19 @@ class PostgreSQLComDescribeExecutorTest {
when(connectionSession.getDatabaseConnectionManager()).thenReturn(databaseConnectionManager);
}
+ private void prepareJDBCBackendConnectionForResultMetaData(final String
sql, final int columnCount) throws SQLException {
+ ProxyDatabaseConnectionManager databaseConnectionManager =
mock(ProxyDatabaseConnectionManager.class);
+ Connection connection = mock(Connection.class);
+ PreparedStatement preparedStatement = mock(PreparedStatement.class);
+ ResultSetMetaData metaData = mock(ResultSetMetaData.class);
+ when(metaData.getColumnCount()).thenReturn(columnCount);
+ when(preparedStatement.getMetaData()).thenReturn(metaData);
+ when(preparedStatement.getConnection()).thenReturn(connection);
+ when(connection.prepareStatement(sql)).thenReturn(preparedStatement);
+ when(databaseConnectionManager.getConnections(any(),
nullable(String.class), anyInt(), anyInt(),
any(ConnectionMode.class))).thenReturn(Collections.singletonList(connection));
+
when(connectionSession.getDatabaseConnectionManager()).thenReturn(databaseConnectionManager);
+ }
+
private void prepareJDBCBackendConnectionWithParamTypes(final String sql,
final int[] paramTypes, final String[] paramTypeNames) throws SQLException {
ParameterMetaData parameterMetaData = mock(ParameterMetaData.class);
for (int i = 0; i < paramTypes.length; i++) {
@@ -738,7 +772,7 @@ class PostgreSQLComDescribeExecutorTest {
expectedColumn("k", Types.INTEGER, 4, "int4"),
expectedColumn("c", Types.CHAR, -1, "char"),
expectedColumn("pad", Types.CHAR, -1, "char"),
- expectedColumn("t_order", Types.VARCHAR, -1, "varchar"),
- expectedColumn("alias_t_order", Types.VARCHAR, -1, "varchar"));
+ new PostgreSQLColumnDescription("t_order", 0, 2249, -1,
PostgreSQLValueFormat.TEXT.getCode()),
+ new PostgreSQLColumnDescription("alias_t_order", 0, 2249, -1,
PostgreSQLValueFormat.TEXT.getCode()));
}
}
diff --git
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutorTest.java
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutorTest.java
index 00cf2bb5984..2c64ed34715 100644
---
a/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutorTest.java
+++
b/proxy/frontend/dialect/postgresql/src/test/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/simple/PostgreSQLComQueryExecutorTest.java
@@ -38,6 +38,7 @@ import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader
import
org.apache.shardingsphere.proxy.backend.response.header.query.QueryResponseHeader;
import
org.apache.shardingsphere.proxy.backend.response.header.update.UpdateResponseHeader;
import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
+import
org.apache.shardingsphere.proxy.backend.postgresql.response.header.query.PostgreSQLQueryHeaderBuilder;
import org.apache.shardingsphere.proxy.frontend.command.executor.ResponseType;
import
org.apache.shardingsphere.proxy.frontend.postgresql.command.PortalContext;
import
org.apache.shardingsphere.sql.parser.statement.core.segment.dal.VariableAssignSegment;
@@ -63,6 +64,7 @@ import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import java.sql.SQLException;
+import java.sql.Types;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -115,7 +117,9 @@ class PostgreSQLComQueryExecutorTest {
@Test
void assertExecuteQueryWithColumnDescription() throws SQLException,
ReflectiveOperationException {
QueryResponseHeader queryResponseHeader =
mock(QueryResponseHeader.class);
-
when(queryResponseHeader.getQueryHeaders()).thenReturn(Collections.singletonList(new
QueryHeader("schema", "table", "label", "column", 1, "type", 2, 3, true, true,
true, true)));
+ when(queryResponseHeader.getQueryHeaders()).thenReturn(
+ Collections.singletonList(new QueryHeader("schema", "table",
"label", "column", Types.STRUCT, "record_type", 2, 3, true, true, true, true,
+
Collections.singletonMap(PostgreSQLQueryHeaderBuilder.TYPE_OID, 2249))));
when(proxyBackendHandler.execute()).thenReturn(queryResponseHeader);
Collection<DatabasePacket> actual = queryExecutor.execute();
PostgreSQLRowDescriptionPacket rowDescriptionPacket =
(PostgreSQLRowDescriptionPacket) actual.iterator().next();
@@ -126,7 +130,7 @@ class PostgreSQLComQueryExecutorTest {
assertThat(columnDescription.getColumnName(), is("label"));
assertThat(columnDescription.getColumnIndex(), is(1));
assertThat(columnDescription.getColumnLength(), is(2));
- assertThat(columnDescription.getTypeOID(), is(new
PostgreSQLColumnDescription("column", 1, 1, 2, "type").getTypeOID()));
+ assertThat(columnDescription.getTypeOID(), is(2249));
assertThat(queryExecutor.getResponseType(), is(ResponseType.QUERY));
}
