Package org.apache.flink.cdc.debezium.internal

Class FlinkDatabaseSchemaHistory

java.lang.Object

org.apache.flink.cdc.debezium.internal.FlinkDatabaseSchemaHistory

All Implemented Interfaces:

io.debezium.relational.history.DatabaseHistory

public class FlinkDatabaseSchemaHistory
extends Object
implements io.debezium.relational.history.DatabaseHistory

The FlinkDatabaseSchemaHistory only stores the latest schema of the monitored tables. When recovering from the checkpoint, it should apply all the tables to the DatabaseSchema, which doesn't need to replay the history anymore.

Considering the data structure maintained in the FlinkDatabaseSchemaHistory is much different from the FlinkDatabaseHistory, it's not compatible with the FlinkDatabaseHistory. Because it only maintains the latest schema of the table rather than all history DDLs, it's useful to prevent OOM when meet massive history DDLs.

Field Summary

Fields

Modifier and Type	Field	Description
static String	DATABASE_HISTORY_INSTANCE_NAME

Fields inherited from interface io.debezium.relational.history.DatabaseHistory

CONFIGURATION_FIELD_PREFIX_STRING, DDL_FILTER, NAME, SKIP_UNPARSEABLE_DDL_STATEMENTS, STORE_ONLY_CAPTURED_TABLES_DDL, STORE_ONLY_MONITORED_TABLES_DDL

Constructor Summary

Constructors

Constructor	Description
FlinkDatabaseSchemaHistory()

Method Summary

Modifier and Type	Method	Description
void	configure(io.debezium.config.Configuration config, io.debezium.relational.history.HistoryRecordComparator comparator, io.debezium.relational.history.DatabaseHistoryListener listener, boolean useCatalogBeforeSchema)
boolean	exists()
void	initializeStorage()
static boolean	isCompatible(Collection<SchemaRecord> records)	Determine whether the FlinkDatabaseSchemaHistory is compatible with the specified state.
void	record(Map<String,?> source, Map<String,?> position, String databaseName, String ddl)
void	record(Map<String,?> source, Map<String,?> position, String databaseName, String schemaName, String ddl, io.debezium.relational.history.TableChanges changes)
void	recover(Map<String,?> source, Map<String,?> position, io.debezium.relational.Tables schema, io.debezium.relational.ddl.DdlParser ddlParser)
void	recover(Map<Map<String,?>,Map<String,?>> offsets, io.debezium.relational.Tables schema, io.debezium.relational.ddl.DdlParser ddlParser)
boolean	skipUnparseableDdlStatements()
void	start()
void	stop()
boolean	storageExists()
boolean	storeOnlyCapturedTables()

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.debezium.relational.history.DatabaseHistory

recover

Field Detail

DATABASE_HISTORY_INSTANCE_NAME

public static final String DATABASE_HISTORY_INSTANCE_NAME

See Also:

Constant Field Values

Constructor Detail

FlinkDatabaseSchemaHistory

public FlinkDatabaseSchemaHistory()

Method Detail

configure

public void configure(io.debezium.config.Configuration config,
                      io.debezium.relational.history.HistoryRecordComparator comparator,
                      io.debezium.relational.history.DatabaseHistoryListener listener,
                      boolean useCatalogBeforeSchema)

Specified by:

configure in interface io.debezium.relational.history.DatabaseHistory

start

public void start()

Specified by:

start in interface io.debezium.relational.history.DatabaseHistory

record

public void record(Map<String,?> source,
                   Map<String,?> position,
                   String databaseName,
                   String ddl)
            throws io.debezium.relational.history.DatabaseHistoryException

Specified by:

record in interface io.debezium.relational.history.DatabaseHistory

Throws:

io.debezium.relational.history.DatabaseHistoryException

record

public void record(Map<String,?> source,
                   Map<String,?> position,
                   String databaseName,
                   String schemaName,
                   String ddl,
                   io.debezium.relational.history.TableChanges changes)
            throws io.debezium.relational.history.DatabaseHistoryException

Specified by:

record in interface io.debezium.relational.history.DatabaseHistory

Throws:

io.debezium.relational.history.DatabaseHistoryException

recover

public void recover(Map<String,?> source,
                    Map<String,?> position,
                    io.debezium.relational.Tables schema,
                    io.debezium.relational.ddl.DdlParser ddlParser)

Specified by:

recover in interface io.debezium.relational.history.DatabaseHistory

recover

public void recover(Map<Map<String,?>,Map<String,?>> offsets,
                    io.debezium.relational.Tables schema,
                    io.debezium.relational.ddl.DdlParser ddlParser)

Specified by:

recover in interface io.debezium.relational.history.DatabaseHistory

stop

public void stop()

Specified by:

stop in interface io.debezium.relational.history.DatabaseHistory

exists

public boolean exists()

Specified by:

exists in interface io.debezium.relational.history.DatabaseHistory

storageExists

public boolean storageExists()

Specified by:

storageExists in interface io.debezium.relational.history.DatabaseHistory

initializeStorage

public void initializeStorage()

Specified by:

initializeStorage in interface io.debezium.relational.history.DatabaseHistory

storeOnlyCapturedTables

public boolean storeOnlyCapturedTables()

Specified by:

storeOnlyCapturedTables in interface io.debezium.relational.history.DatabaseHistory

skipUnparseableDdlStatements

public boolean skipUnparseableDdlStatements()

Specified by:

skipUnparseableDdlStatements in interface io.debezium.relational.history.DatabaseHistory

isCompatible

public static boolean isCompatible(Collection<SchemaRecord> records)

Determine whether the FlinkDatabaseSchemaHistory is compatible with the specified state.