Package io.debezium.relational

Class RelationalSnapshotChangeEventSource<P extends Partition,O extends OffsetContext>

java.lang.Object

io.debezium.pipeline.source.AbstractSnapshotChangeEventSource<P,O>

io.debezium.relational.RelationalSnapshotChangeEventSource<P,O>

All Implemented Interfaces:

ChangeEventSource, SnapshotChangeEventSource<P,O>, AutoCloseable

public abstract class RelationalSnapshotChangeEventSource<P extends Partition,O extends OffsetContext>
extends AbstractSnapshotChangeEventSource<P,O>

Base class for SnapshotChangeEventSource for relational databases with or without a schema history.

A transaction is managed by this base class, sub-classes shouldn't rollback or commit this transaction. They are free to use nested transactions or savepoints, though.

Author:

Gunnar Morling

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
(package private) static interface	RelationalSnapshotChangeEventSource.CheckedFunction<T,R>	Functional interface for operations that may throw SQLException.
(package private) static interface	RelationalSnapshotChangeEventSource.PooledWork<T extends OffsetContext>	Functional interface for snapshot work that uses pooled resources.
private static final record	RelationalSnapshotChangeEventSource.PreparedTables	Holds the prepared table information for snapshot processing.
static class	RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P extends Partition,O extends OffsetContext>	Mutable context which is populated in the course of snapshotting.
private static class	RelationalSnapshotChangeEventSource.SnapshotMarkerResolver	Determines the appropriate SnapshotRecord marker based on record position.
private static final record	RelationalSnapshotChangeEventSource.SnapshotSelect
private static class	RelationalSnapshotChangeEventSource.ThreadedSnapshotExecutor	Manages threaded execution of snapshot work using a thread pool.

Nested classes/interfaces inherited from class io.debezium.pipeline.source.AbstractSnapshotChangeEventSource

AbstractSnapshotChangeEventSource.SnapshotContext<P extends Partition,O extends OffsetContext>

Nested classes/interfaces inherited from interface io.debezium.pipeline.source.spi.ChangeEventSource

ChangeEventSource.ChangeEventSourceContext

Field Summary

Fields

Modifier and Type	Field	Description
protected final io.debezium.util.Clock	clock
protected Queue<JdbcConnection>	connectionPool
private final RelationalDatabaseConnectorConfig	connectorConfig
protected final EventDispatcher<P,TableId>	dispatcher
private final JdbcConnection	jdbcConnection
private final MainConnectionProvidingConnectionFactory<? extends JdbcConnection>	jdbcConnectionFactory
private static final org.slf4j.Logger	LOGGER
static final Pattern	MATCH_ALL_PATTERN
private final RelationalDatabaseSchema	schema
static final Pattern	SELECT_ALL_PATTERN
private final TableId	signalDataCollectionTableId
private final SnapshotProgressListener<P>	snapshotProgressListener
protected final SnapshotterService	snapshotterService

Fields inherited from class io.debezium.pipeline.source.AbstractSnapshotChangeEventSource

LOG_INTERVAL, notificationService

Constructor Summary

Constructors

Constructor

Description

RelationalSnapshotChangeEventSource(RelationalDatabaseConnectorConfig connectorConfig, MainConnectionProvidingConnectionFactory<? extends JdbcConnection> jdbcConnectionFactory, RelationalDatabaseSchema schema, EventDispatcher<P,TableId> dispatcher, io.debezium.util.Clock clock, SnapshotProgressListener<P> snapshotProgressListener, NotificationService<P,O> notificationService, SnapshotterService snapshotterService)

Method Summary

Modifier and Type	Method	Description
protected boolean	additionalColumnFilter(P partition, TableId tableId, String columnName)	Additional filter handling for preparing column names for snapshot select
private Set<TableId>	addSignalingCollectionAndSort(Set<TableId> capturedTables)
protected int	calculateChunkCount(OptionalLong rowCount, int maxThreads, int multiplier)
protected void	connectionCreated(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)	Executes steps which have to be taken just after the database connection is created.
protected void	connectionPoolConnectionCreated(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, JdbcConnection connection)	Executes steps which have to be taken just after a connection pool connection is created.
protected abstract O	copyOffset(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
private void	createChunkedDataEvents(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, Queue<JdbcConnection> connectionPool, Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable, Queue<O> offsets, EventDispatcher.SnapshotReceiver<P> snapshotReceiver)
private Queue<JdbcConnection>	createConnectionPool(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> ctx)
private void	createDataEvents(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, Queue<JdbcConnection> connectionPool, Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable)
protected Callable<Void>	createDataEventsForChunkedTableCallable(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, EventDispatcher.SnapshotReceiver<P> snapshotReceiver, SnapshotChunk chunk, Map<TableId,TableChunkProgress> progressMap, SnapshotProgress snapshotProgress, Queue<JdbcConnection> connectionPool, Queue<O> offsets)
protected Callable<Void>	createDataEventsForTableCallable(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, EventDispatcher.SnapshotReceiver<P> snapshotReceiver, Table table, boolean firstTable, boolean lastTable, int tableOrder, int tableCount, String selectStatement, OptionalLong rowCount, Set<TableId> rowCountTablesKeySet, Queue<JdbcConnection> connectionPool, Queue<O> offsets)
private void	createLegacyDataEvents(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, Queue<JdbcConnection> connectionPool, Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable, Queue<O> offsets, EventDispatcher.SnapshotReceiver<P> snapshotReceiver)
private Queue<O>	createOffsetPool(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, int poolSize)
private Callable<Void>	createPooledResourceCallable(Queue<JdbcConnection> connectionPool, Queue<O> offsetPool, RelationalSnapshotChangeEventSource.PooledWork<O> work, Runnable errorHandler)	Creates a Callable that borrows resources from pools, executes work, and returns resources.
protected void	createSchemaChangeEventsForTables(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, SnapshottingTask snapshottingTask)
Connection	createSnapshotConnection()
private void	determineCapturedTables(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> ctx, Set<Pattern> dataCollectionsToBeSnapshotted, SnapshottingTask snapshottingTask)
protected abstract void	determineSnapshotOffset(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, O previousOffset)	Determines the current offset (MySQL binlog position, Oracle SCN etc.), storing it into the passed context object.
private RelationalSnapshotChangeEventSource.SnapshotSelect	determineSnapshotSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, TableId tableId, Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable)	Returns a valid query string for the specified table, either given by the user via snapshot select overrides or defaulting to a statement provided by the DB-specific change event source.
protected void	doCreateDataEventsForChunk(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, O offset, EventDispatcher.SnapshotReceiver<P> snapshotReceiver, SnapshotChunk chunk, Map<TableId,TableChunkProgress> progressMap, SnapshotProgress snapshotProgress, JdbcConnection jdbcConnection)
protected void	doCreateDataEventsForTable(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, O offset, EventDispatcher.SnapshotReceiver<P> snapshotReceiver, Table table, boolean firstTable, boolean lastTable, int tableOrder, int tableCount, String selectStatement, OptionalLong rowCount, Set<TableId> rowCountTablesKeySet, JdbcConnection jdbcConnection)
SnapshotResult<O>	doExecute(ChangeEventSource.ChangeEventSourceContext context, O previousOffset, AbstractSnapshotChangeEventSource.SnapshotContext<P,O> snapshotContext, SnapshottingTask snapshottingTask)	Executes this source.
private void	emitRecordWithCoordination(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, O offset, EventDispatcher.SnapshotReceiver<P> snapshotReceiver, SnapshotChunk chunk, TableChunkProgress progress, SnapshotProgress snapshotProgress, TableId tableId, Object[] row, Instant sourceTableSnapshotTimestamp, boolean isFirstRecord, boolean isLastRecord)	Emits a record with proper coordination to ensure correct snapshot marker ordering.
protected String	enhanceOverriddenSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, String overriddenSelect, TableId tableId)	This method is overridden for Oracle to implement "as of SCN" predicate
protected abstract Set<TableId>	getAllTableIds(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)	Returns all candidate tables; the current filter configuration will be applied to the result set, resulting in the effective set of captured tables.
SnapshottingTask	getBlockingSnapshottingTask(P partition, O previousOffset, SnapshotConfiguration snapshotConfiguration)	Returns the blocking snapshotting task based on the snapshot configuration from the signal.
protected ChangeRecordEmitter<P>	getChangeRecordEmitter(P partition, O offset, TableId tableId, Object[] row, Instant timestamp)	Returns a ChangeRecordEmitter producing the change records for the given table row.
protected io.debezium.util.Clock	getClock()
protected abstract SchemaChangeEvent	getCreateTableEvent(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, Table table)	Creates a SchemaChangeEvent representing the creation of the given table.
protected List<Column>	getKeyColumnsForChunking(Table table)
protected List<String>	getPreparedColumnNames(P partition, Table table)	Prepares a list of columns to be used in the snapshot select.
protected List<Pattern>	getSignalDataCollectionPattern(String signalingDataCollection)
protected Optional<String>	getSnapshotConnectionFirstSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, TableId tableId)
protected abstract Optional<String>	getSnapshotSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, TableId tableId, List<String> columns)	Returns the SELECT statement to be used for scanning the given table or empty value if the table will be streamed from but not snapshotted
protected String	getSnapshotSelectOverridesByTable(TableId tableId, Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverrides)
protected Instant	getSnapshotSourceTimestamp(JdbcConnection jdbcConnection, O offset, TableId tableId)	For the given table gets source.ts_ms value from the database for snapshot data!
SnapshottingTask	getSnapshottingTask(P partition, O previousOffset)	Returns the snapshotting task based on the previous offset (if available) and the connector's snapshotting mode.
private io.debezium.util.Threads.Timer	getTableScanLogTimer()
protected Collection<TableId>	getTablesForSchemaChange(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
private void	handleEmptyChunkCoordination(SnapshotChunk chunk, TableChunkProgress progress, SnapshotProgress snapshotProgress)	Handles coordination for empty chunks to ensure latches are properly signaled.
private void	initializePooledConnection(JdbcConnection connection)
private boolean	isUseNonChunkedSnapshots(int snapshotMaxThreads)
protected void	lastSnapshotRecord(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
protected abstract void	lockTablesForSchemaSnapshot(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)	Locks all tables to be captured, so that no concurrent schema changes can be applied to them.
protected void	postSnapshot()
private RelationalSnapshotChangeEventSource.PreparedTables	prepareTables(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, Function<TableId,RelationalSnapshotChangeEventSource.SnapshotSelect> selectGenerator, RelationalSnapshotChangeEventSource.CheckedFunction<TableId,OptionalLong> rowCountProvider)	Prepares tables for snapshot by generating select statements and row counts.
protected void	preSnapshot()
protected Statement	readTableStatement(JdbcConnection jdbcConnection, OptionalLong tableSize)	Allow per-connector query creation to override for best database performance depending on the table size.
protected abstract void	readTableStructure(ChangeEventSource.ChangeEventSourceContext sourceContext, RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext, O offsetContext, SnapshottingTask snapshottingTask)	Reads the structure of all the captured tables, writing it to RelationalSnapshotChangeEventSource.RelationalSnapshotContext.tables.
protected void	releaseDataSnapshotLocks(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)	Releases all locks established in order to create a consistent data snapshot.
protected abstract void	releaseSchemaSnapshotLocks(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)	Releases all locks established in order to create a consistent schema snapshot.
protected ResultSet	resultSetForDataEvents(String selectStatement, Statement statement)
private void	rollbackTransaction(Connection connection)
protected OptionalLong	rowCountForTable(TableId tableId)	If connector is able to provide statistics-based number of records per table.
protected Long	rowCountForTableChunked(TableId tableId)
private void	setChunkSnapshotMarker(OffsetContext offset, SnapshotChunk chunk, boolean firstRecordInChunk, boolean lastRecordInChunk)
private void	setSnapshotMarker(OffsetContext offset, boolean firstTable, boolean lastTable, boolean firstRecordInTable, boolean lastRecordInTable)
private Map<TableId,OptionalLong>	sortByRowCount(Map<TableId,OptionalLong> rowCountTables, RelationalDatabaseConnectorConfig.SnapshotTablesRowCountOrder order)
private Stream<TableId>	toTableIds(Set<TableId> tableIds, Pattern pattern)
protected void	tryStartingSnapshot(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)

Methods inherited from class io.debezium.pipeline.source.AbstractSnapshotChangeEventSource

aborted, close, completed, delaySnapshotIfNeeded, determineDataCollectionsToBeSnapshotted, execute, getDataCollectionPattern, getOffsets, prepare

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

LOGGER

private static final org.slf4j.Logger LOGGER

SELECT_ALL_PATTERN

public static final Pattern SELECT_ALL_PATTERN

MATCH_ALL_PATTERN

public static final Pattern MATCH_ALL_PATTERN

connectorConfig

private final RelationalDatabaseConnectorConfig connectorConfig

jdbcConnection

private final JdbcConnection jdbcConnection

jdbcConnectionFactory

private final MainConnectionProvidingConnectionFactory<? extends JdbcConnection> jdbcConnectionFactory

schema

private final RelationalDatabaseSchema schema

dispatcher

protected final EventDispatcher<P extends Partition,TableId> dispatcher

clock

protected final io.debezium.util.Clock clock

snapshotProgressListener

private final SnapshotProgressListener<P extends Partition> snapshotProgressListener

snapshotterService

protected final SnapshotterService snapshotterService

connectionPool

protected Queue<JdbcConnection> connectionPool

signalDataCollectionTableId

private final TableId signalDataCollectionTableId

Constructor Details

RelationalSnapshotChangeEventSource

public RelationalSnapshotChangeEventSource(RelationalDatabaseConnectorConfig connectorConfig,
 MainConnectionProvidingConnectionFactory<? extends JdbcConnection> jdbcConnectionFactory,
 RelationalDatabaseSchema schema,
 EventDispatcher<P,TableId> dispatcher,
 io.debezium.util.Clock clock,
 SnapshotProgressListener<P> snapshotProgressListener,
 NotificationService<P,O> notificationService,
 SnapshotterService snapshotterService)

Method Details

doExecute

public SnapshotResult<O> doExecute(ChangeEventSource.ChangeEventSourceContext context,
 O previousOffset,
 AbstractSnapshotChangeEventSource.SnapshotContext<P,O> snapshotContext,
 SnapshottingTask snapshottingTask)
                            throws Exception

Description copied from class: AbstractSnapshotChangeEventSource

Executes this source. Implementations should regularly check via the given context if they should stop. If that's the case, they should abort their processing and perform any clean-up needed, such as rolling back pending transactions, releasing locks, etc.

Specified by:

doExecute in class AbstractSnapshotChangeEventSource<P extends Partition,O extends OffsetContext>

Parameters:

context - contextual information for this source's execution

previousOffset - previous offset restored from Kafka

snapshotContext - mutable context information populated throughout the snapshot process

snapshottingTask - immutable information about what tasks should be performed during snapshot

Returns:

an indicator to the position at which the snapshot was taken

Throws:

Exception

createConnectionPool

private Queue<JdbcConnection> createConnectionPool(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> ctx)
                                            throws SQLException

Throws:

SQLException

initializePooledConnection

private void initializePooledConnection(JdbcConnection connection)
                                 throws SQLException

Throws:

SQLException

createSnapshotConnection

public Connection createSnapshotConnection()
                                    throws SQLException

Throws:

SQLException

getBlockingSnapshottingTask

public SnapshottingTask getBlockingSnapshottingTask(P partition,
 O previousOffset,
 SnapshotConfiguration snapshotConfiguration)

Description copied from interface: SnapshotChangeEventSource

Returns the blocking snapshotting task based on the snapshot configuration from the signal.

getSnapshottingTask

public SnapshottingTask getSnapshottingTask(P partition,
 O previousOffset)

Description copied from interface: SnapshotChangeEventSource

Returns the snapshotting task based on the previous offset (if available) and the connector's snapshotting mode.

connectionCreated

protected void connectionCreated(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
                          throws Exception

Executes steps which have to be taken just after the database connection is created.

Throws:

Exception

connectionPoolConnectionCreated

protected void connectionPoolConnectionCreated(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 JdbcConnection connection)
                                        throws SQLException

Executes steps which have to be taken just after a connection pool connection is created.

Throws:

SQLException

getSignalDataCollectionPattern

protected List<Pattern> getSignalDataCollectionPattern(String signalingDataCollection)

toTableIds

private Stream<TableId> toTableIds(Set<TableId> tableIds,
 Pattern pattern)

addSignalingCollectionAndSort

private Set<TableId> addSignalingCollectionAndSort(Set<TableId> capturedTables)

determineCapturedTables

private void determineCapturedTables(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> ctx,
 Set<Pattern> dataCollectionsToBeSnapshotted,
 SnapshottingTask snapshottingTask)
                              throws Exception

Throws:

Exception

getAllTableIds

protected abstract Set<TableId> getAllTableIds(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
                                        throws Exception

Returns all candidate tables; the current filter configuration will be applied to the result set, resulting in the effective set of captured tables.

Throws:

Exception

lockTablesForSchemaSnapshot

protected abstract void lockTablesForSchemaSnapshot(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
                                             throws Exception

Locks all tables to be captured, so that no concurrent schema changes can be applied to them.

Throws:

Exception

determineSnapshotOffset

protected abstract void determineSnapshotOffset(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 O previousOffset)
                                         throws Exception

Determines the current offset (MySQL binlog position, Oracle SCN etc.), storing it into the passed context object. Subsequently, the DB's schema (and data) will be be read at this position. Once the snapshot is completed, a StreamingChangeEventSource will be set up with this initial position to continue with stream reading from there.

Throws:

Exception

readTableStructure

protected abstract void readTableStructure(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 O offsetContext,
 SnapshottingTask snapshottingTask)
                                    throws Exception

Reads the structure of all the captured tables, writing it to RelationalSnapshotChangeEventSource.RelationalSnapshotContext.tables.

Throws:

Exception

releaseSchemaSnapshotLocks

protected abstract void releaseSchemaSnapshotLocks(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
                                            throws Exception

Releases all locks established in order to create a consistent schema snapshot.

Throws:

Exception

releaseDataSnapshotLocks

protected void releaseDataSnapshotLocks(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)
                                 throws Exception

Releases all locks established in order to create a consistent data snapshot.

Throws:

Exception

createSchemaChangeEventsForTables

protected void createSchemaChangeEventsForTables(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 SnapshottingTask snapshottingTask)
                                          throws Exception

Throws:

Exception

getTablesForSchemaChange

protected Collection<TableId> getTablesForSchemaChange(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)

getCreateTableEvent

protected abstract SchemaChangeEvent getCreateTableEvent(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 Table table)
                                                  throws Exception

Creates a SchemaChangeEvent representing the creation of the given table.

Throws:

Exception

createDataEvents

private void createDataEvents(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 Queue<JdbcConnection> connectionPool,
 Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable)
                       throws Exception

Throws:

Exception

isUseNonChunkedSnapshots

private boolean isUseNonChunkedSnapshots(int snapshotMaxThreads)

createLegacyDataEvents

private void createLegacyDataEvents(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 Queue<JdbcConnection> connectionPool,
 Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable,
 Queue<O> offsets,
 EventDispatcher.SnapshotReceiver<P> snapshotReceiver)
                             throws Exception

Throws:

Exception

createChunkedDataEvents

private void createChunkedDataEvents(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 Queue<JdbcConnection> connectionPool,
 Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable,
 Queue<O> offsets,
 EventDispatcher.SnapshotReceiver<P> snapshotReceiver)
                              throws Exception

Throws:

Exception

emitRecordWithCoordination

private void emitRecordWithCoordination(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 O offset,
 EventDispatcher.SnapshotReceiver<P> snapshotReceiver,
 SnapshotChunk chunk,
 TableChunkProgress progress,
 SnapshotProgress snapshotProgress,
 TableId tableId,
 Object[] row,
 Instant sourceTableSnapshotTimestamp,
 boolean isFirstRecord,
 boolean isLastRecord)
                                 throws InterruptedException

Emits a record with proper coordination to ensure correct snapshot marker ordering.

Throws:

InterruptedException

handleEmptyChunkCoordination

private void handleEmptyChunkCoordination(SnapshotChunk chunk,
 TableChunkProgress progress,
 SnapshotProgress snapshotProgress)
                                   throws InterruptedException

Handles coordination for empty chunks to ensure latches are properly signaled.

Throws:

InterruptedException

getKeyColumnsForChunking

protected List<Column> getKeyColumnsForChunking(Table table)

calculateChunkCount

protected int calculateChunkCount(OptionalLong rowCount,
 int maxThreads,
 int multiplier)

createOffsetPool

private Queue<O> createOffsetPool(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 int poolSize)

sortByRowCount

private Map<TableId,OptionalLong> sortByRowCount(Map<TableId,OptionalLong> rowCountTables,
 RelationalDatabaseConnectorConfig.SnapshotTablesRowCountOrder order)

copyOffset

protected abstract O copyOffset(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)

tryStartingSnapshot

protected void tryStartingSnapshot(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)

getSnapshotSourceTimestamp

protected Instant getSnapshotSourceTimestamp(JdbcConnection jdbcConnection,
 O offset,
 TableId tableId)

For the given table gets source.ts_ms value from the database for snapshot data! For Postgresql its globally static for all tables since postgresql snapshot process setting auto commit off. For Mysql its static per table and might be ~second behind of the select statements start ts.

createDataEventsForTableCallable

protected Callable<Void> createDataEventsForTableCallable(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 EventDispatcher.SnapshotReceiver<P> snapshotReceiver,
 Table table,
 boolean firstTable,
 boolean lastTable,
 int tableOrder,
 int tableCount,
 String selectStatement,
 OptionalLong rowCount,
 Set<TableId> rowCountTablesKeySet,
 Queue<JdbcConnection> connectionPool,
 Queue<O> offsets)

createDataEventsForChunkedTableCallable

protected Callable<Void> createDataEventsForChunkedTableCallable(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 EventDispatcher.SnapshotReceiver<P> snapshotReceiver,
 SnapshotChunk chunk,
 Map<TableId,TableChunkProgress> progressMap,
 SnapshotProgress snapshotProgress,
 Queue<JdbcConnection> connectionPool,
 Queue<O> offsets)

doCreateDataEventsForTable

protected void doCreateDataEventsForTable(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 O offset,
 EventDispatcher.SnapshotReceiver<P> snapshotReceiver,
 Table table,
 boolean firstTable,
 boolean lastTable,
 int tableOrder,
 int tableCount,
 String selectStatement,
 OptionalLong rowCount,
 Set<TableId> rowCountTablesKeySet,
 JdbcConnection jdbcConnection)
                                   throws InterruptedException,
SQLException

Throws:

InterruptedException

SQLException

doCreateDataEventsForChunk

protected void doCreateDataEventsForChunk(ChangeEventSource.ChangeEventSourceContext sourceContext,
 RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 O offset,
 EventDispatcher.SnapshotReceiver<P> snapshotReceiver,
 SnapshotChunk chunk,
 Map<TableId,TableChunkProgress> progressMap,
 SnapshotProgress snapshotProgress,
 JdbcConnection jdbcConnection)
                                   throws InterruptedException,
SQLException

Throws:

InterruptedException

SQLException

resultSetForDataEvents

protected ResultSet resultSetForDataEvents(String selectStatement,
 Statement statement)
                                    throws SQLException

Throws:

SQLException

setSnapshotMarker

private void setSnapshotMarker(OffsetContext offset,
 boolean firstTable,
 boolean lastTable,
 boolean firstRecordInTable,
 boolean lastRecordInTable)

setChunkSnapshotMarker

private void setChunkSnapshotMarker(OffsetContext offset,
 SnapshotChunk chunk,
 boolean firstRecordInChunk,
 boolean lastRecordInChunk)

lastSnapshotRecord

protected void lastSnapshotRecord(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext)

rowCountForTable

protected OptionalLong rowCountForTable(TableId tableId)

If connector is able to provide statistics-based number of records per table.

rowCountForTableChunked

protected Long rowCountForTableChunked(TableId tableId)
                                throws SQLException

Throws:

SQLException

getTableScanLogTimer

private io.debezium.util.Threads.Timer getTableScanLogTimer()

getChangeRecordEmitter

protected ChangeRecordEmitter<P> getChangeRecordEmitter(P partition,
 O offset,
 TableId tableId,
 Object[] row,
 Instant timestamp)

Returns a ChangeRecordEmitter producing the change records for the given table row.

determineSnapshotSelect

private RelationalSnapshotChangeEventSource.SnapshotSelect determineSnapshotSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 TableId tableId,
 Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverridesByTable)

Returns a valid query string for the specified table, either given by the user via snapshot select overrides or defaulting to a statement provided by the DB-specific change event source.

Parameters:

tableId - the table to generate a query for

snapshotSelectOverridesByTable - the select overrides by table

Returns:

a snapshot select record, never null valid query string or empty if table will not be snapshotted

getSnapshotSelectOverridesByTable

protected String getSnapshotSelectOverridesByTable(TableId tableId,
 Map<io.debezium.spi.schema.DataCollectionId,String> snapshotSelectOverrides)

getPreparedColumnNames

protected List<String> getPreparedColumnNames(P partition,
 Table table)

Prepares a list of columns to be used in the snapshot select. The selected columns are based on the column include/exclude filters and if all columns are excluded, the list will contain all the primary key columns.

Returns:

list of snapshot select columns

additionalColumnFilter

protected boolean additionalColumnFilter(P partition,
 TableId tableId,
 String columnName)

Additional filter handling for preparing column names for snapshot select

enhanceOverriddenSelect

protected String enhanceOverriddenSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 String overriddenSelect,
 TableId tableId)

This method is overridden for Oracle to implement "as of SCN" predicate

Parameters:

snapshotContext - snapshot context, used for getting offset SCN

overriddenSelect - conditional snapshot select

Returns:

enhanced select statement. By default it just returns original select statements.

getSnapshotSelect

protected abstract Optional<String> getSnapshotSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 TableId tableId,
 List<String> columns)

Returns the SELECT statement to be used for scanning the given table or empty value if the table will be streamed from but not snapshotted

getSnapshotConnectionFirstSelect

protected Optional<String> getSnapshotConnectionFirstSelect(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 TableId tableId)

readTableStatement

protected Statement readTableStatement(JdbcConnection jdbcConnection,
 OptionalLong tableSize)
                                throws SQLException

Allow per-connector query creation to override for best database performance depending on the table size.

Throws:

SQLException

rollbackTransaction

private void rollbackTransaction(Connection connection)

prepareTables

private RelationalSnapshotChangeEventSource.PreparedTables prepareTables(RelationalSnapshotChangeEventSource.RelationalSnapshotContext<P,O> snapshotContext,
 Function<TableId,RelationalSnapshotChangeEventSource.SnapshotSelect> selectGenerator,
 RelationalSnapshotChangeEventSource.CheckedFunction<TableId,OptionalLong> rowCountProvider)
                                                                  throws SQLException

Prepares tables for snapshot by generating select statements and row counts.

Parameters:

snapshotContext - the snapshot context

selectGenerator - function to generate select statement for a table

rowCountProvider - function to get row count for a table

Returns:

the prepared tables with query and row count maps

Throws:

SQLException - if row count retrieval fails

createPooledResourceCallable

private Callable<Void> createPooledResourceCallable(Queue<JdbcConnection> connectionPool,
 Queue<O> offsetPool,
 RelationalSnapshotChangeEventSource.PooledWork<O> work,
 Runnable errorHandler)

Creates a Callable that borrows resources from pools, executes work, and returns resources.

Parameters:

connectionPool - the pool of JDBC connections

offsetPool - the pool of offset contexts

work - the work to execute with borrowed resources

errorHandler - called if work throws an exception, can be null

Returns:

a Callable wrapping the resource management

getClock

protected io.debezium.util.Clock getClock()

postSnapshot

protected void postSnapshot()
                     throws InterruptedException

Throws:

InterruptedException

preSnapshot

protected void preSnapshot()
                    throws InterruptedException

Throws:

InterruptedException