Skip to content

Schema refresh fix INC-8256 #262

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yashisrivastava-confluent merged 4 commits into from
Jan 28, 2026
Merged

Schema refresh fix INC-8256 #262

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1e7959c
Removing extra schema refresh calls
yashisrivastava-confluent Jan 16, 2026
8df45c2
Add fix
naveenmall11 Jan 16, 2026
f42ff03
Add fix
naveenmall11 Jan 17, 2026
8cfd66e
Remove extra debug logs
yashisrivastava-confluent Jan 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions ...res/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,6 @@ protected void connectionCreated(RelationalSnapshotContext<PostgresPartition, Po
if (snapshotterService.getSnapshotter().shouldStreamEventsStartingFromSnapshot() && startingSlotInfo == null) {
setSnapshotTransactionIsolationLevel(snapshotContext.onDemand);
}
schema.refresh(jdbcConnection, false);
}

@Override
Expand Down Expand Up @@ -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);
}
}

Expand Down
8 changes: 3 additions & 5 deletions ...es/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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();
Copy link
Member

@akanimesh7 akanimesh7 Jan 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yashisrivastava-confluent For my understanding, init schema before starting streaming is not required, because streaming starts exactly from the point noted during snapshot. Is it ?

Copy link
Member

@yashisrivastava-confluent yashisrivastava-confluent Jan 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes


// 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)) {
Expand Down
2 changes: 1 addition & 1 deletion debezium-core/src/main/java/io/debezium/jdbc/JdbcConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down