You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2026/07/27 06:26:23 UTC
(shardingsphere) branch master updated: Fix PostgreSQL composite column type OID resolution (#39241)
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 <zh...@apache.org>
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));
}