Fix PostgreSQL composite column type OID resolution

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)

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 @@
 /**
  * Column description for PostgreSQL.
  */
+@RequiredArgsConstructor
 @Getter
 public final class PostgreSQLColumnDescription {
 

proxy/frontend/dialect/postgresql/src/main/java/org/apache/shardingsphere/proxy/frontend/postgresql/command/query/PostgreSQLColumnTypeOIDLoader.java

@@ -0,0 +1,120 @@
+/*
+ * 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.frontend.postgresql.command.query;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.apache.shardingsphere.infra.connection.kernel.KernelProcessor;
+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.session.query.QueryContext;
+import org.apache.shardingsphere.proxy.backend.response.header.query.QueryHeader;
+import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
+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.List;
+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 the routed backend connection.
+     *
+     * @param connectionSession connection session
+     * @param queryContext query context
+     * @param queryHeaders query headers
+     * @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 ConnectionSession connectionSession, final QueryContext queryContext, final List<QueryHeader> queryHeaders) throws SQLException {
+        if (queryHeaders.stream().noneMatch(each -> Types.STRUCT == each.getColumnType())) {
+            return Collections.emptyMap();
+        }
+        ExecutionContext executionContext = new KernelProcessor().generateExecutionContext(queryContext, queryContext.getMetaData().getGlobalRuleMetaData(), queryContext.getMetaData().getProps());
+        if (1 != executionContext.getExecutionUnits().size()) {
+            return Collections.emptyMap();
+        }
+        ExecutionUnit executionUnit = executionContext.getExecutionUnits().iterator().next();
+        List<Connection> connections = connectionSession.getDatabaseConnectionManager().getConnections(
+                connectionSession.getUsedDatabaseName(), executionUnit.getDataSourceName(), 0, 1, ConnectionMode.CONNECTION_STRICTLY);
+        return load(connections.get(0), queryHeaders);
+    }
+
+    private static Map<Integer, Integer> load(final Connection connection, final List<QueryHeader> queryHeaders) throws SQLException {
+        return connection.isWrapperFor(BaseConnection.class) ? getTypeOIDs(connection.unwrap(BaseConnection.class), queryHeaders) : Collections.emptyMap();
+    }
+
+    /**
+     * 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) ? getTypeOIDs(connection.unwrap(BaseConnection.class), metaData) : Collections.emptyMap();
+    }
+
+    private static Map<Integer, Integer> getTypeOIDs(final BaseConnection connection, final List<QueryHeader> queryHeaders) throws SQLException {
+        Map<Integer, Integer> result = new HashMap<>();
+        for (int i = 0; i < queryHeaders.size(); i++) {
+            QueryHeader each = queryHeaders.get(i);
+            Optional<Integer> typeOID = findTypeOID(connection, each.getColumnType(), each.getColumnTypeName());
+            if (typeOID.isPresent()) {
+                result.put(i + 1, typeOID.get());
+            }
+        }
+        return result;
+    }
+
+    private static Map<Integer, Integer> getTypeOIDs(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++) {
+            Optional<Integer> typeOID = findTypeOID(connection, metaData.getColumnType(columnIndex), metaData.getColumnTypeName(columnIndex));
+            if (typeOID.isPresent()) {
+                result.put(columnIndex, typeOID.get());
+            }
+        }
+        return result;
+    }
+
+    private static Optional<Integer> findTypeOID(final BaseConnection connection, final int columnType, final String columnTypeName) throws SQLException {
+        if (Types.STRUCT == columnType) {
+            int typeOID = connection.getTypeInfo().getPGType(columnTypeName);
+            if (Oid.UNSPECIFIED != typeOID) {
+                return Optional.of(typeOID);
+            }
+        }
+        return Optional.empty();
+    }
+}

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.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.frontend.postgresql.command.query.PostgreSQLColumnTypeOIDLoader;
 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;
@@ -61,6 +62,7 @@
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 
 /**
  * PostgreSQL portal.
@@ -81,8 +83,12 @@ public final class Portal {
 
     private final ProxyBackendHandler proxyBackendHandler;
 
+    private final QueryContext queryContext;
+
     private final ProxyDatabaseConnectionManager databaseConnectionManager;
 
+    private Map<Integer, Integer> columnTypeOIDs;
+
     private ResponseHeader responseHeader;
 
     public Portal(final String name, final PostgreSQLServerPreparedStatement preparedStatement, final List<Object> params, final List<PostgreSQLValueFormat> resultFormats,
@@ -97,7 +103,7 @@ public Portal(final String name, final PostgreSQLServerPreparedStatement prepare
             ((ParameterAware) sqlStatementContext).bindParameters(params);
         }
         DatabaseType protocolType = ProxyContext.getInstance().getContextManager().getDatabase(databaseName).getProtocolType();
-        QueryContext queryContext = new QueryContext(sqlStatementContext, preparedStatement.getSql(), params, preparedStatement.getHintValueContext(),
+        queryContext = new QueryContext(sqlStatementContext, preparedStatement.getSql(), params, preparedStatement.getHintValueContext(),
                 databaseConnectionManager.getConnectionSession().getConnectionContext(), ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData(), true);
         proxyBackendHandler = ProxyBackendHandlerFactory.newInstance(protocolType, queryContext, databaseConnectionManager.getConnectionSession(), true);
     }
@@ -109,6 +115,10 @@ public Portal(final String name, final PostgreSQLServerPreparedStatement prepare
      */
     public void bind() throws SQLException {
         responseHeader = proxyBackendHandler.execute();
+        if (responseHeader instanceof QueryResponseHeader) {
+            QueryResponseHeader queryResponseHeader = (QueryResponseHeader) responseHeader;
+            columnTypeOIDs = PostgreSQLColumnTypeOIDLoader.load(databaseConnectionManager.getConnectionSession(), queryContext, queryResponseHeader.getQueryHeaders());
+        }
     }
 
     /**
@@ -136,7 +146,11 @@ private Collection<PostgreSQLColumnDescription> createColumnDescriptions(final Q
         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 = columnTypeOIDs.get(currentColumnIndex);
+            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;
     }