diff --git a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java
index 649096df585..d6eba4d24d6 100644
--- a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java
+++ b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java
@@ -106,7 +106,6 @@ protected void connectionCreated(RelationalSnapshotContext<PostgresPartition, Po
         if (snapshotterService.getSnapshotter().shouldStreamEventsStartingFromSnapshot() && startingSlotInfo == null) {
             setSnapshotTransactionIsolationLevel(snapshotContext.onDemand);
         }
-        schema.refresh(jdbcConnection, false);
     }
 
     @Override
@@ -141,8 +140,6 @@ protected void lockTablesForSchemaSnapshot(ChangeEventSourceContext sourceContex
 
             LOGGER.info("Waiting a maximum of '{}' seconds for each table lock", lockTimeout.getSeconds());
             jdbcConnection.executeWithoutCommitting(statements.toString());
-            // now that we have the locks, refresh the schema
-            schema.refresh(jdbcConnection, false);
         }
     }
 
diff --git a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java
index 5c205fee94a..53cfb84ec2e 100644
--- a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java
+++ b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java
@@ -120,12 +120,14 @@ public void init(PostgresOffsetContext offsetContext) {
     }
 
     private void initSchema() {
+        LOGGER.info("Performing initial schema load");
         try {
             taskContext.refreshSchema(connection, true);
         }
         catch (SQLException e) {
             throw new DebeziumException("Error while executing initial schema load", e);
         }
+        LOGGER.info("Completed initial schema load");
     }
 
     @Override
@@ -156,15 +158,11 @@ public void execute(ChangeEventSourceContext context, PostgresPartition partitio
                 walPosition = new WalPositionLocator();
                 replicationStream.compareAndSet(null, replicationConnection.startStreaming(walPosition));
             }
-            // for large dbs, the refresh of schema can take too much time
-            // such that the connection times out. We must enable keep
-            // alive to ensure that it doesn't time out
+            // Start keep alive thread to prevent connection timeout during time-consuming operations the DB side.
             ReplicationStream stream = this.replicationStream.get();
             stream.startKeepAlive(
                     Threads.newSingleThreadExecutor(PostgresConnector.class, connectorConfig.getLogicalName(), KEEP_ALIVE_THREAD_NAME, threadNameContext));
 
-            initSchema();
-
             // If we need to do a pre-snapshot streaming catch up, we should allow the snapshot transaction to persist
             // but normally we want to start streaming without any open transactions.
             if (!isInPreSnapshotCatchUpStreaming(this.effectiveOffset)) {
diff --git a/debezium-core/src/main/java/io/debezium/jdbc/JdbcConnection.java b/debezium-core/src/main/java/io/debezium/jdbc/JdbcConnection.java
index ffac694d915..57cd435f453 100644
--- a/debezium-core/src/main/java/io/debezium/jdbc/JdbcConnection.java
+++ b/debezium-core/src/main/java/io/debezium/jdbc/JdbcConnection.java
@@ -1251,7 +1251,7 @@ public void readSchema(Tables tables, String databaseCatalog, String schemaNameP
             for (TableId includeTable : tableIds) {
                 LOGGER.debug("Retrieving columns of table {}", includeTable);
 
-                Map<TableId, List<Column>> cols = getColumnsDetails(databaseCatalog, schemaNamePattern, includeTable.table(), tableFilter,
+                Map<TableId, List<Column>> cols = getColumnsDetails(databaseCatalog, includeTable.schema(), includeTable.table(), tableFilter,
                         columnFilter, metadata, viewIds);
                 columnsByTable.putAll(cols);
             }