Package org.jumpmind.symmetric.service.impl

Class DataService

java.lang.Object

org.jumpmind.symmetric.service.impl.AbstractService

org.jumpmind.symmetric.service.impl.DataService

All Implemented Interfaces:

IDataService, IService

public class DataService
extends AbstractService
implements IDataService

See Also:

• IDataService

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
class	DataService.DataMapper
static class	DataService.LastCaptureByChannelMapper
protected class	DataService.TableReloadRequestMapper
protected class	DataService.TableReloadStatusMapper

Field Summary

Fields

Modifier and Type	Field	Description
static final DateTimeFormatter	isoJsonDateTimeFormatter
protected Map<IHeartbeatListener,Long>	lastHeartbeatTimestamps
static final int	PROGRESS_LOG_UPDATE_DELAY_MS
static final int	RECAPTURE_DATA_COMMIT_LIMIT
static final String	TIMESTAMP_ISO_JSON_FORMAT

Fields inherited from class org.jumpmind.symmetric.service.impl.AbstractService

log, parameterService, platform, sqlTemplate, sqlTemplateDirty, symmetricDialect, tablePrefix

Constructor Summary

Constructors

Constructor	Description
DataService(ISymmetricEngine engine, IExtensionService extensionService)

Method Summary

Modifier and Type	Method	Description
int	cancelTableReloadRequest(TableReloadRequest request)
protected void	checkInterrupted()
List<TableReloadRequest>	collapseTableReloadRequestsByLoadId(List<TableReloadRequest> requests)
protected void	convertDataToReload(Data data, Table table, TriggerHistory hist, String nodeId)
protected Map<String,TableReloadRequest>	convertReloadListToMap(List<TableReloadRequest> reloadRequests, List<TriggerRouter> triggerRouters, boolean isFullLoad, boolean isChannelLoad)
int	countData()
long	countDataGaps()
int	countDataInRange(long firstDataId, long secondDataId)	Count the number of data ids in a range
Data	createData(String catalogName, String schemaName, String tableName)
Data	createData(String catalogName, String schemaName, String tableName, String whereClause)
Data	createData(ISqlTransaction transaction, String catalogName, String schemaName, String tableName, String whereClause)
protected Data	createData(ISqlTransaction transaction, Trigger trigger, String whereClause)
protected void	createPurgeEvent(ISqlTransaction transaction, String sql, Node targetNode, Node sourceNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, boolean isLoad, long loadId, String createBy)
protected void	createPurgeEvent(ISqlTransaction transaction, String sql, Node targetNode, Node sourceNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, boolean isLoad, long loadId, String createBy, AbstractBatch.Status outgoingBatchStatus)
void	createTableReloadStatus(ISqlTransaction transaction, long loadId, boolean isFullLoad, String sourceNodeId, String targetNodeId)
void	deleteAllDataGaps(ISqlTransaction transaction)
void	deleteCapturedConfigChannelData()
void	deleteDataGap(ISqlTransaction transaction, DataGap gap)
void	deleteDataGap(DataGap gap)
void	deleteDataGaps(ISqlTransaction transaction, Collection<DataGap> gaps)
protected void	deleteTableReloadRequest(ISqlTransaction sqlTransaction, TableReloadRequest request)
void	expireDataGaps(ISqlTransaction transaction, Collection<DataGap> gaps)
protected Data	fetchRecapturedData(Data data, String recaptureTransactionId)	Attempts to recapture expired data row, which is still relevant and assigns a new transaction ID
Date	findCreateTimeOfData(long dataId)
Date	findCreateTimeOfEvent(long dataId)
Data	findData(long dataId)
List<Data>	findData(long startDataId, long endDataId)
List<DataGap>	findDataGaps()
protected List<DataGap>	findDataGaps(boolean isExpired)
List<DataGap>	findDataGapsExpired()
List<DataGap>	findDataGapsUnchecked()
long	findMaxDataEventDataId()
long	findMaxDataId()
long	findMinDataId()
Date	findNextCreateTimeOfDataStartingAt(long dataId)
String	findNodeIdsByNodeGroupId()
boolean	fixLastDataGap()
String	generateTransactionIdFromTimestamp(String prefix, Instant instant)	Helper.
List<TableReloadStatus>	getActiveIncomingTableReloadStatus()
List<TableReloadStatus>	getActiveOutgoingTableReloadStatus()
List<TableReloadStatus>	getActiveTableReloadStatus()
protected String	getCsvDataFor(ISqlTransaction transaction, Trigger trigger, TriggerHistory triggerHistory, String whereClause, boolean pkOnly)
protected long	getDataCountForReload(Table table, Node targetNode, String selectSql)
ISqlRowMapper<Data>	getDataMapper()
protected String	getDataOrderBy()
protected String	getDataSelectByBatchSql(long batchId, long startDataId, String channelId)
protected String	getDataSelectSql(long batchId, long startDataId, boolean isContainsBigLob)
protected String	getDataSelectSql(long batchId, long startDataId, String channelId)
protected List<IHeartbeatListener>	getHeartbeatListeners(boolean force)	Get a list of IHeartbeatListeners that are ready for a heartbeat according to IHeartbeatListener.getTimeBetweenHeartbeatsInSeconds()
List<TableReloadStatus>	getIncomingTableReloadStatus()
Map<String,Date>	getLastDataCaptureByChannel()
List<TableReloadStatus>	getOutgoingTableReloadStatus()
protected String	getReloadChannelIdForTrigger(Trigger trigger, Map<String,Channel> channels)
TableReloadRequest	getTableReloadRequest(long loadId)
TableReloadRequest	getTableReloadRequest(long loadId, String triggerId, String routerId)
TableReloadRequest	getTableReloadRequest(TableReloadRequestKey key)
List<TableReloadRequest>	getTableReloadRequestByLoadId()
Map<Long,List<TableReloadRequest>>	getTableReloadRequestByLoadIdMap()
List<TableReloadRequest>	getTableReloadRequests()
List<TableReloadRequest>	getTableReloadRequests(long loadId)
List<TableReloadRequest>	getTableReloadRequestToProcess(String sourceNodeId)
List<TableReloadRequest>	getTableReloadRequestToProcessByTarget(String targetNodeId)
List<TableReloadStatus>	getTableReloadStatus()
TableReloadStatus	getTableReloadStatusByLoadIdAndSourceNodeId(long loadId, String sourceNodeId)
List<TableReloadStatus>	getTableReloadStatusByTarget(String targetNodeId)
protected int	getTransformMultiplier(Table table, TriggerRouter triggerRouter)
protected boolean	hasColumnDataIntegrity(Data data, TriggerHistory hist)
void	heartbeat(boolean force)	Update Node information for this node and call IHeartbeatListeners.
protected void	insertCreateEvent(ISqlTransaction transaction, Node targetNode, TriggerHistory triggerHistory, boolean isLoad, long loadId, String createBy, boolean excludeIndices, boolean excludeForeignKeys, boolean excludeDefaults)
void	insertCreateEvent(ISqlTransaction transaction, Node targetNode, TriggerHistory triggerHistory, String channelId, boolean isLoad, long loadId, String createBy, boolean excludeIndices, boolean excludeForeignKeys, boolean excludeDefaults)
void	insertCreateEvent(ISqlTransaction transaction, Node targetNode, TriggerHistory triggerHistory, String channelId, boolean isLoad, long loadId, String createBy, boolean excludeIndices, boolean excludeForeignKeys, boolean excludeDefaults, AbstractBatch.Status outgoingBatchStatus)
void	insertCreateEvent(Node targetNode, TriggerHistory triggerHistory, boolean isLoad, long loadId, String createBy, boolean excludeIndices, boolean excludeForeignKeys, boolean excludeDefaults)
void	insertCreateEvent(Node targetNode, TriggerHistory triggerHistory, String createBy, boolean excludeIndices, boolean excludeForeignKeys, boolean excludeDefaults)
long	insertData(ISqlTransaction transaction, Data data)
long	insertData(Data data)
long	insertDataAndDataEventAndOutgoingBatch(ISqlTransaction transaction, Data data, String nodeId, boolean isLoad, long loadId, String createBy, AbstractBatch.Status status, long estimatedBatchRowCount)
long	insertDataAndDataEventAndOutgoingBatch(ISqlTransaction transaction, Data data, String nodeId, boolean isLoad, long loadId, String createBy, AbstractBatch.Status status, String overrideChannelId, long estimatedBatchRowCount)
long	insertDataAndDataEventAndOutgoingBatch(Data data, String nodeId, boolean isLoad, long loadId, String createBy)
void	insertDataAndDataEventAndOutgoingBatch(Data data, String channelId, List<Node> nodes, boolean isLoad, long loadId, String createBy)
protected void	insertDataEvent(ISqlTransaction transaction, long dataId, long batchId)
protected void	insertDataEvent(ISqlTransaction transaction, DataEvent dataEvent)
protected long	insertDataEventAndOutgoingBatch(ISqlTransaction transaction, long dataId, String channelId, String nodeId, DataEventType eventType, boolean isLoad, long loadId, String createBy, AbstractBatch.Status status, String tableName, long estimatedBatchRowCount)
void	insertDataEvents(ISqlTransaction transaction, List<DataEvent> events)
void	insertDataGap(ISqlTransaction transaction, DataGap gap)
void	insertDataGap(DataGap gap)
void	insertDataGaps(ISqlTransaction transaction, Collection<DataGap> gaps)
void	insertHeartbeatEvent(Node node, boolean isReload)	Because we can't add a trigger on the _node table, we are artificially generating heartbeat events.
protected void	insertPurgeEvent(ISqlTransaction transaction, Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, boolean isLoad, String overrideDeleteStatement, long loadId, String createBy)
protected void	insertPurgeEvent(ISqlTransaction transaction, Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, boolean isLoad, String overrideDeleteStatement, long loadId, String createBy, AbstractBatch.Status outgoingBatchStatus)
protected int	insertRecapturedData(List<Data> insertList)	Inserts recaptured data back into sym_data
long	insertReloadEvent(ISqlTransaction transaction, Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, String overrideInitialLoadSelect, boolean isLoad, long loadId, String createBy, AbstractBatch.Status status, long estimatedBatchRowCount)
long	insertReloadEvent(ISqlTransaction transaction, Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, String overrideInitialLoadSelect, boolean isLoad, long loadId, String createBy, AbstractBatch.Status status, String channelId, long estimatedBatchRowCount)
protected long	insertReloadEvent(ISqlTransaction transaction, Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, String overrideInitialLoadSelect, boolean isLoad, long loadId, String createBy, AbstractBatch.Status status, String channelId, long estimatedBatchRowCount, boolean isImmediate)
boolean	insertReloadEvent(TableReloadRequest request, boolean deleteAtClient)
long	insertReloadEventImmediate(ISqlTransaction transaction, Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, String overrideInitialLoadSelect, boolean isLoad, long loadId, String createBy, AbstractBatch.Status status, String channelId, long estimatedBatchRowCount)
Map<Integer,ExtractRequest>	insertReloadEvents(Node targetNode, boolean reverse, List<TableReloadRequest> reloadRequests, ProcessInfo processInfo, List<TriggerRouter> triggerRouters, Map<Integer,ExtractRequest> extractRequests, IReloadGenerator reloadGenerator)
protected long	insertRequestedOutgoingBatches(ISqlTransaction transaction, Node targetNode, TriggerRouter triggerRouter, TriggerHistory triggerHistory, String overrideInitialLoadSelect, long loadId, String createBy, String channelId, long totalRows, long maxRowsPerBatch, long batchCount)
void	insertScriptEvent(String channelId, Node targetNode, String script, boolean isLoad, long loadId, String createBy)
void	insertScriptEvent(ISqlTransaction transaction, String channelId, Node targetNode, String script, boolean isLoad, long loadId, String createBy)
void	insertSqlEvent(ISqlTransaction transaction, Node targetNode, String sql, boolean isLoad, long loadId, String createBy)
void	insertSqlEvent(ISqlTransaction transaction, Node targetNode, String sql, boolean isLoad, long loadId, String createBy, AbstractBatch.Status outgoingBatchStatus)
void	insertSqlEvent(ISqlTransaction transaction, TriggerHistory history, String channelId, Node targetNode, String sql, boolean isLoad, long loadId, String createBy)
void	insertSqlEvent(ISqlTransaction transaction, TriggerHistory history, String channelId, Node targetNode, String sql, boolean isLoad, long loadId, String createBy, AbstractBatch.Status outgoingBatchStatus)
void	insertSqlEvent(Node targetNode, String sql, boolean isLoad, long loadId, String createBy)
void	insertTableReloadRequest(ISqlTransaction transaction, TableReloadRequest request)
void	insertTableReloadRequest(TableReloadRequest request)
protected boolean	isLocalNodeUsingBulkLoad(String nodeId)
List<Data>	listData(long batchId, String nodeId, long startDataId, String channelId, int maxRowsToRetrieve)
List<Number>	listDataIds(long batchId, String nodeId)
int	reCaptureData(long minDataId, long maxDataId)	Attempts to recapture stale data rows between specified Ids and re-insert them back into sym_data with a new transaction ID.
protected int	reCaptureData(List<Data> dataList)	Attempts to recapture stale data rows and re-insert them back into sym_data with a new transaction ID (all records are in one transaction).
protected String[]	recaptureKeysForData(Table table, Data data)	Looks up values of primary keys for specified data row object
protected String	recaptureWhereFilterForKeys(Table table, TriggerHistory hist, String[] keys, Object[] values)	Extracts filter for a where clause for specified values of primary keys
void	reloadMissingForeignKeyRows(long batchId, String nodeId, long dataId, long rowNumber)
protected void	reloadMissingForeignKeyRows(Data data, long batchId, String nodeId, long dataId, long rowNumber)
void	reloadMissingForeignKeyRowsForLoad(String sourceNodeId, long batchId, long rowNumber, Table table, CsvData data, String channelId)
void	reloadMissingForeignKeyRowsReverse(String sourceNodeId, Table table, CsvData data, String channelId, boolean sendCorrectionToPeers)
String	reloadNode(String nodeId, boolean reverseLoad, String createBy)
String	reloadTable(String nodeId, String catalogName, String schemaName, String tableName)
String	reloadTable(String nodeId, String catalogName, String schemaName, String tableName, String overrideInitialLoadSelect)
protected String	reloadTable(String nodeId, String catalogName, String schemaName, String tableName, String overrideInitialLoadSelect, String overrideChannelId, boolean isImmediate)
String	reloadTableImmediate(String nodeId, String catalogName, String schemaName, String tableName, String overrideInitialLoadSelect, String overrideChannelId)
int	resendBatchAsReload(long batchId, String nodeId)
int	resendDataAsReload(long minDataId, long maxDataId)
protected void	resendDataAsReload(List<Data> dataList, String nodeId)
List<String>	resolveTargetTables(String sql, TriggerRouter triggerRouter, TriggerHistory triggerHistory, Node targetNode)
ISqlReadCursor<Data>	selectDataFor(Long batchId, String channelId)
ISqlReadCursor<Data>	selectDataFor(Long batchId, String targetNodeId, boolean isContainsBigLob)
ISqlReadCursor<Data>	selectDataFor(Batch batch)
void	sendMissingForeignKeyRowsForLoad(long batchId, String nodeId, long rowNumber, String rowData)
void	sendNewerDataToNode(ISqlTransaction transaction, String targetNodeId, String tableName, String pkCsvData, Date minCreateTime, String winningNodeId)
boolean	sendSchema(String nodeId, String catalogName, String schemaName, String tableName, boolean isLoad, boolean excludeIndices, boolean excludeForeignKeys, boolean excludeDefaults)
void	sendScript(String nodeId, String script, boolean isLoad)
String	sendSQL(String nodeId, String sql)
String	sendSQL(String nodeId, String catalogName, String schemaName, String tableName, String sql)	Sends a SQL command to the remote node for execution by creating a SQL event that is synced like other data
protected void	updateLastHeartbeatTime(List<IHeartbeatListener> listeners)
int	updateTableReloadRequestsCancelled(long loadId, String sourceNodeId)
protected int	updateTableReloadRequestsError(long loadId, String sourceNodeId, SqlException e)
void	updateTableReloadRequestsLoadId(ISqlTransaction transaction, long loadId, TableReloadRequest request)
void	updateTableReloadStatusDataCounts(ISqlTransaction transaction, long loadId, String sourceNodeId, long startBatchId, long endBatchId, long dataBatchCount, long rowsCount)
TableReloadStatus	updateTableReloadStatusDataLoaded(ISqlTransaction transaction, long loadId, String sourceNodeId, long batchId, int batchCount, boolean isBulkLoaded)
void	updateTableReloadStatusFailed(ISqlTransaction transaction, long loadId, String sourceNodeId, long batchId)
void	updateTableReloadStatusSetupCount(ISqlTransaction transaction, long loadId, String sourceNodeId, int setupBatchCount)
void	updateTableReloadStatusTableCount(ISqlTransaction transaction, long loadId, String sourceNodeId, int tableCount)
protected void	validate(TableReloadRequest reloadRequest, List<TriggerRouter> triggerRouters)

Methods inherited from class org.jumpmind.symmetric.service.impl.AbstractService

assertNotNull, buildBatchOrderBy, buildBatchParams, buildBatchWhere, buildBatchWhereFromFilter, close, createSqlReplacementTokens, getJdbcTemplate, getParameterService, getSql, getSymmetricDialect, getTablePrefix, getTargetDialect, getTargetPlatform, getTargetPlatform, isCalledFromSymmetricAdminTool, isSet, isStreamClosedByClient, isSymmetricTable, logOnce, maxDate, readAcks, sendAck, setSqlMap, synchronize, toNodeIds, toNodeIds

Methods inherited from class java.lang.Object

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

Field Details

RECAPTURE_DATA_COMMIT_LIMIT

public static final int RECAPTURE_DATA_COMMIT_LIMIT

See Also:

• Constant Field Values

PROGRESS_LOG_UPDATE_DELAY_MS

public static final int PROGRESS_LOG_UPDATE_DELAY_MS

See Also:

• Constant Field Values

TIMESTAMP_ISO_JSON_FORMAT

public static final transient String TIMESTAMP_ISO_JSON_FORMAT

See Also:

• Constant Field Values

isoJsonDateTimeFormatter

public static final transient DateTimeFormatter isoJsonDateTimeFormatter

lastHeartbeatTimestamps

protected Map<IHeartbeatListener,Long> lastHeartbeatTimestamps

Constructor Details

DataService

public DataService(ISymmetricEngine engine,
 IExtensionService extensionService)

Method Details

insertReloadEvent

public boolean insertReloadEvent(TableReloadRequest request,
 boolean deleteAtClient)

generateTransactionIdFromTimestamp

public String generateTransactionIdFromTimestamp(String prefix,
 Instant instant)

Helper. Creates an ISO-compliant transaction Id string for specified instant and prefix.

cancelTableReloadRequest

public int cancelTableReloadRequest(TableReloadRequest request)

Specified by:

cancelTableReloadRequest in interface IDataService

deleteTableReloadRequest

protected void deleteTableReloadRequest(ISqlTransaction sqlTransaction,
 TableReloadRequest request)

insertTableReloadRequest

public void insertTableReloadRequest(TableReloadRequest request)

Specified by:

insertTableReloadRequest in interface IDataService

insertTableReloadRequest

public void insertTableReloadRequest(ISqlTransaction transaction,
 TableReloadRequest request)

Specified by:

insertTableReloadRequest in interface IDataService

getTableReloadRequest

public TableReloadRequest getTableReloadRequest(TableReloadRequestKey key)

Specified by:

getTableReloadRequest in interface IDataService

getTableReloadRequest

public TableReloadRequest getTableReloadRequest(long loadId)

Specified by:

getTableReloadRequest in interface IDataService

getTableReloadRequests

public List<TableReloadRequest> getTableReloadRequests(long loadId)

Specified by:

getTableReloadRequests in interface IDataService

getTableReloadRequest

public TableReloadRequest getTableReloadRequest(long loadId,
 String triggerId,
 String routerId)

Specified by:

getTableReloadRequest in interface IDataService

getTableReloadRequestToProcess

public List<TableReloadRequest> getTableReloadRequestToProcess(String sourceNodeId)

Specified by:

getTableReloadRequestToProcess in interface IDataService

getTableReloadRequestToProcessByTarget

public List<TableReloadRequest> getTableReloadRequestToProcessByTarget(String targetNodeId)

Specified by:

getTableReloadRequestToProcessByTarget in interface IDataService

getTableReloadRequests

public List<TableReloadRequest> getTableReloadRequests()

getTableReloadStatus

public List<TableReloadStatus> getTableReloadStatus()

Specified by:

getTableReloadStatus in interface IDataService

getOutgoingTableReloadStatus

public List<TableReloadStatus> getOutgoingTableReloadStatus()

Specified by:

getOutgoingTableReloadStatus in interface IDataService

getIncomingTableReloadStatus

public List<TableReloadStatus> getIncomingTableReloadStatus()

Specified by:

getIncomingTableReloadStatus in interface IDataService

getActiveTableReloadStatus

public List<TableReloadStatus> getActiveTableReloadStatus()

Specified by:

getActiveTableReloadStatus in interface IDataService

getActiveOutgoingTableReloadStatus

public List<TableReloadStatus> getActiveOutgoingTableReloadStatus()

Specified by:

getActiveOutgoingTableReloadStatus in interface IDataService

getActiveIncomingTableReloadStatus

public List<TableReloadStatus> getActiveIncomingTableReloadStatus()

Specified by:

getActiveIncomingTableReloadStatus in interface IDataService

getTableReloadStatusByLoadIdAndSourceNodeId

public TableReloadStatus getTableReloadStatusByLoadIdAndSourceNodeId(long loadId,
 String sourceNodeId)

Specified by:

getTableReloadStatusByLoadIdAndSourceNodeId in interface IDataService

getTableReloadStatusByTarget

public List<TableReloadStatus> getTableReloadStatusByTarget(String targetNodeId)

Specified by:

getTableReloadStatusByTarget in interface IDataService

getTableReloadRequestByLoadId

public List<TableReloadRequest> getTableReloadRequestByLoadId()

collapseTableReloadRequestsByLoadId

public List<TableReloadRequest> collapseTableReloadRequestsByLoadId(List<TableReloadRequest> requests)

getTableReloadRequestByLoadIdMap

public Map<Long,List<TableReloadRequest>> getTableReloadRequestByLoadIdMap()

Specified by:

getTableReloadRequestByLoadIdMap in interface IDataService

updateTableReloadStatusDataLoaded

public TableReloadStatus updateTableReloadStatusDataLoaded(ISqlTransaction transaction,
 long loadId,
 String sourceNodeId,
 long batchId,
 int batchCount,
 boolean isBulkLoaded)

Specified by:

updateTableReloadStatusDataLoaded in interface IDataService

updateTableReloadStatusFailed

public void updateTableReloadStatusFailed(ISqlTransaction transaction,
 long loadId,
 String sourceNodeId,
 long batchId)

Specified by:

updateTableReloadStatusFailed in interface IDataService

updateTableReloadStatusDataCounts

public void updateTableReloadStatusDataCounts(ISqlTransaction transaction,
 long loadId,
 String sourceNodeId,
 long startBatchId,
 long endBatchId,
 long dataBatchCount,
 long rowsCount)

updateTableReloadRequestsLoadId

public void updateTableReloadRequestsLoadId(ISqlTransaction transaction,
 long loadId,
 TableReloadRequest request)

updateTableReloadStatusTableCount

public void updateTableReloadStatusTableCount(ISqlTransaction transaction,
 long loadId,
 String sourceNodeId,
 int tableCount)

createTableReloadStatus

public void createTableReloadStatus(ISqlTransaction transaction,
 long loadId,
 boolean isFullLoad,
 String sourceNodeId,
 String targetNodeId)

updateTableReloadStatusSetupCount

public void updateTableReloadStatusSetupCount(ISqlTransaction transaction,
 long loadId,
 String sourceNodeId,
 int setupBatchCount)

updateTableReloadRequestsCancelled

public int updateTableReloadRequestsCancelled(long loadId,
 String sourceNodeId)

Specified by:

updateTableReloadRequestsCancelled in interface IDataService

updateTableReloadRequestsError

protected int updateTableReloadRequestsError(long loadId,
 String sourceNodeId,
 SqlException e)

insertRequestedOutgoingBatches

protected long insertRequestedOutgoingBatches(ISqlTransaction transaction,
 Node targetNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 String overrideInitialLoadSelect,
 long loadId,
 String createBy,
 String channelId,
 long totalRows,
 long maxRowsPerBatch,
 long batchCount)

insertReloadEvent

public long insertReloadEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 String overrideInitialLoadSelect,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status status,
 long estimatedBatchRowCount)

Specified by:

insertReloadEvent in interface IDataService

Returns:

If isLoad then return the inserted batch id otherwise return the data id

insertReloadEvent

public long insertReloadEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 String overrideInitialLoadSelect,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status status,
 String channelId,
 long estimatedBatchRowCount)

Parameters:

estimatedBatchRowCount - TODO

Returns:

If isLoad then return the inserted batch id otherwise return the data id

insertReloadEventImmediate

public long insertReloadEventImmediate(ISqlTransaction transaction,
 Node targetNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 String overrideInitialLoadSelect,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status status,
 String channelId,
 long estimatedBatchRowCount)

insertReloadEvent

protected long insertReloadEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 String overrideInitialLoadSelect,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status status,
 String channelId,
 long estimatedBatchRowCount,
 boolean isImmediate)

getReloadChannelIdForTrigger

protected String getReloadChannelIdForTrigger(Trigger trigger,
 Map<String,Channel> channels)

insertReloadEvents

public Map<Integer,ExtractRequest> insertReloadEvents(Node targetNode,
 boolean reverse,
 List<TableReloadRequest> reloadRequests,
 ProcessInfo processInfo,
 List<TriggerRouter> triggerRouters,
 Map<Integer,ExtractRequest> extractRequests,
 IReloadGenerator reloadGenerator)

Specified by:

insertReloadEvents in interface IDataService

convertReloadListToMap

protected Map<String,TableReloadRequest> convertReloadListToMap(List<TableReloadRequest> reloadRequests,
 List<TriggerRouter> triggerRouters,
 boolean isFullLoad,
 boolean isChannelLoad)

validate

protected void validate(TableReloadRequest reloadRequest,
 List<TriggerRouter> triggerRouters)

getDataCountForReload

protected long getDataCountForReload(Table table,
 Node targetNode,
 String selectSql)
                              throws SqlException

Throws:

SqlException

getTransformMultiplier

protected int getTransformMultiplier(Table table,
 TriggerRouter triggerRouter)

isLocalNodeUsingBulkLoad

protected boolean isLocalNodeUsingBulkLoad(String nodeId)

insertPurgeEvent

protected void insertPurgeEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 boolean isLoad,
 String overrideDeleteStatement,
 long loadId,
 String createBy)

insertPurgeEvent

protected void insertPurgeEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 boolean isLoad,
 String overrideDeleteStatement,
 long loadId,
 String createBy,
 AbstractBatch.Status outgoingBatchStatus)

resolveTargetTables

public List<String> resolveTargetTables(String sql,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 Node targetNode)

createPurgeEvent

protected void createPurgeEvent(ISqlTransaction transaction,
 String sql,
 Node targetNode,
 Node sourceNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 boolean isLoad,
 long loadId,
 String createBy)

createPurgeEvent

protected void createPurgeEvent(ISqlTransaction transaction,
 String sql,
 Node targetNode,
 Node sourceNode,
 TriggerRouter triggerRouter,
 TriggerHistory triggerHistory,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status outgoingBatchStatus)

insertSqlEvent

public void insertSqlEvent(Node targetNode,
 String sql,
 boolean isLoad,
 long loadId,
 String createBy)

Specified by:

insertSqlEvent in interface IDataService

insertSqlEvent

public void insertSqlEvent(ISqlTransaction transaction,
 Node targetNode,
 String sql,
 boolean isLoad,
 long loadId,
 String createBy)

Specified by:

insertSqlEvent in interface IDataService

insertSqlEvent

public void insertSqlEvent(ISqlTransaction transaction,
 Node targetNode,
 String sql,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status outgoingBatchStatus)

insertSqlEvent

public void insertSqlEvent(ISqlTransaction transaction,
 TriggerHistory history,
 String channelId,
 Node targetNode,
 String sql,
 boolean isLoad,
 long loadId,
 String createBy)

Specified by:

insertSqlEvent in interface IDataService

insertSqlEvent

public void insertSqlEvent(ISqlTransaction transaction,
 TriggerHistory history,
 String channelId,
 Node targetNode,
 String sql,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status outgoingBatchStatus)

insertScriptEvent

public void insertScriptEvent(String channelId,
 Node targetNode,
 String script,
 boolean isLoad,
 long loadId,
 String createBy)

Specified by:

insertScriptEvent in interface IDataService

insertScriptEvent

public void insertScriptEvent(ISqlTransaction transaction,
 String channelId,
 Node targetNode,
 String script,
 boolean isLoad,
 long loadId,
 String createBy)

Specified by:

insertScriptEvent in interface IDataService

countDataInRange

public int countDataInRange(long firstDataId,
 long secondDataId)

Description copied from interface: IDataService

Count the number of data ids in a range

Specified by:

countDataInRange in interface IDataService

countData

public int countData()

Specified by:

countData in interface IDataService

insertCreateEvent

public void insertCreateEvent(Node targetNode,
 TriggerHistory triggerHistory,
 boolean isLoad,
 long loadId,
 String createBy,
 boolean excludeIndices,
 boolean excludeForeignKeys,
 boolean excludeDefaults)

Specified by:

insertCreateEvent in interface IDataService

insertCreateEvent

public void insertCreateEvent(Node targetNode,
 TriggerHistory triggerHistory,
 String createBy,
 boolean excludeIndices,
 boolean excludeForeignKeys,
 boolean excludeDefaults)

Specified by:

insertCreateEvent in interface IDataService

insertCreateEvent

protected void insertCreateEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerHistory triggerHistory,
 boolean isLoad,
 long loadId,
 String createBy,
 boolean excludeIndices,
 boolean excludeForeignKeys,
 boolean excludeDefaults)

insertCreateEvent

public void insertCreateEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerHistory triggerHistory,
 String channelId,
 boolean isLoad,
 long loadId,
 String createBy,
 boolean excludeIndices,
 boolean excludeForeignKeys,
 boolean excludeDefaults)

Specified by:

insertCreateEvent in interface IDataService

insertCreateEvent

public void insertCreateEvent(ISqlTransaction transaction,
 Node targetNode,
 TriggerHistory triggerHistory,
 String channelId,
 boolean isLoad,
 long loadId,
 String createBy,
 boolean excludeIndices,
 boolean excludeForeignKeys,
 boolean excludeDefaults,
 AbstractBatch.Status outgoingBatchStatus)

insertData

public long insertData(Data data)

Specified by:

insertData in interface IDataService

insertData

public long insertData(ISqlTransaction transaction,
 Data data)

Specified by:

insertData in interface IDataService

insertDataEvent

protected void insertDataEvent(ISqlTransaction transaction,
 DataEvent dataEvent)

insertDataEvent

protected void insertDataEvent(ISqlTransaction transaction,
 long dataId,
 long batchId)

insertDataEvents

public void insertDataEvents(ISqlTransaction transaction,
 List<DataEvent> events)

Specified by:

insertDataEvents in interface IDataService

insertDataAndDataEventAndOutgoingBatch

public void insertDataAndDataEventAndOutgoingBatch(Data data,
 String channelId,
 List<Node> nodes,
 boolean isLoad,
 long loadId,
 String createBy)

Specified by:

insertDataAndDataEventAndOutgoingBatch in interface IDataService

insertDataAndDataEventAndOutgoingBatch

public long insertDataAndDataEventAndOutgoingBatch(Data data,
 String nodeId,
 boolean isLoad,
 long loadId,
 String createBy)

Specified by:

insertDataAndDataEventAndOutgoingBatch in interface IDataService

Returns:

The inserted batch id

insertDataAndDataEventAndOutgoingBatch

public long insertDataAndDataEventAndOutgoingBatch(ISqlTransaction transaction,
 Data data,
 String nodeId,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status status,
 String overrideChannelId,
 long estimatedBatchRowCount)

Parameters:

estimatedBatchRowCount - TODO

Returns:

The inserted batch id

insertDataAndDataEventAndOutgoingBatch

public long insertDataAndDataEventAndOutgoingBatch(ISqlTransaction transaction,
 Data data,
 String nodeId,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status status,
 long estimatedBatchRowCount)

Specified by:

insertDataAndDataEventAndOutgoingBatch in interface IDataService

insertDataEventAndOutgoingBatch

protected long insertDataEventAndOutgoingBatch(ISqlTransaction transaction,
 long dataId,
 String channelId,
 String nodeId,
 DataEventType eventType,
 boolean isLoad,
 long loadId,
 String createBy,
 AbstractBatch.Status status,
 String tableName,
 long estimatedBatchRowCount)

reloadNode

public String reloadNode(String nodeId,
 boolean reverseLoad,
 String createBy)

Specified by:

reloadNode in interface IDataService

sendScript

public void sendScript(String nodeId,
 String script,
 boolean isLoad)

Specified by:

sendScript in interface IDataService

sendSchema

public boolean sendSchema(String nodeId,
 String catalogName,
 String schemaName,
 String tableName,
 boolean isLoad,
 boolean excludeIndices,
 boolean excludeForeignKeys,
 boolean excludeDefaults)

Specified by:

sendSchema in interface IDataService

sendSQL

public String sendSQL(String nodeId,
 String catalogName,
 String schemaName,
 String tableName,
 String sql)

Description copied from interface: IDataService

Sends a SQL command to the remote node for execution by creating a SQL event that is synced like other data

Specified by:

sendSQL in interface IDataService

Parameters:

nodeId - the remote node where the SQL statement will be executed

catalogName - used to find the sym_trigger entry for table that will be associated with this event

schemaName - used to find the sym_trigger entry for table that will be associated with this event

tableName - used to find the sym_trigger entry for table that will be associated with this event

sql - the SQL statement to run on the remote node database

Returns:

message string indicating success or error

sendSQL

public String sendSQL(String nodeId,
 String sql)

Specified by:

sendSQL in interface IDataService

reloadTable

public String reloadTable(String nodeId,
 String catalogName,
 String schemaName,
 String tableName)

Specified by:

reloadTable in interface IDataService

reloadTable

public String reloadTable(String nodeId,
 String catalogName,
 String schemaName,
 String tableName,
 String overrideInitialLoadSelect)

Specified by:

reloadTable in interface IDataService

reloadTableImmediate

public String reloadTableImmediate(String nodeId,
 String catalogName,
 String schemaName,
 String tableName,
 String overrideInitialLoadSelect,
 String overrideChannelId)

Specified by:

reloadTableImmediate in interface IDataService

reloadTable

protected String reloadTable(String nodeId,
 String catalogName,
 String schemaName,
 String tableName,
 String overrideInitialLoadSelect,
 String overrideChannelId,
 boolean isImmediate)

reloadMissingForeignKeyRowsReverse

public void reloadMissingForeignKeyRowsReverse(String sourceNodeId,
 Table table,
 CsvData data,
 String channelId,
 boolean sendCorrectionToPeers)

Specified by:

reloadMissingForeignKeyRowsReverse in interface IDataService

reloadMissingForeignKeyRowsForLoad

public void reloadMissingForeignKeyRowsForLoad(String sourceNodeId,
 long batchId,
 long rowNumber,
 Table table,
 CsvData data,
 String channelId)

Specified by:

reloadMissingForeignKeyRowsForLoad in interface IDataService

sendMissingForeignKeyRowsForLoad

public void sendMissingForeignKeyRowsForLoad(long batchId,
 String nodeId,
 long rowNumber,
 String rowData)

reloadMissingForeignKeyRows

public void reloadMissingForeignKeyRows(long batchId,
 String nodeId,
 long dataId,
 long rowNumber)

Specified by:

reloadMissingForeignKeyRows in interface IDataService

reloadMissingForeignKeyRows

protected void reloadMissingForeignKeyRows(Data data,
 long batchId,
 String nodeId,
 long dataId,
 long rowNumber)

sendNewerDataToNode

public void sendNewerDataToNode(ISqlTransaction transaction,
 String targetNodeId,
 String tableName,
 String pkCsvData,
 Date minCreateTime,
 String winningNodeId)

Specified by:

sendNewerDataToNode in interface IDataService

insertHeartbeatEvent

public void insertHeartbeatEvent(Node node,
 boolean isReload)

Because we can't add a trigger on the _node table, we are artificially generating heartbeat events.

Specified by:

insertHeartbeatEvent in interface IDataService

Parameters:

node -

createData

public Data createData(String catalogName,
 String schemaName,
 String tableName)

Specified by:

createData in interface IDataService

createData

public Data createData(String catalogName,
 String schemaName,
 String tableName,
 String whereClause)

Specified by:

createData in interface IDataService

createData

public Data createData(ISqlTransaction transaction,
 String catalogName,
 String schemaName,
 String tableName,
 String whereClause)

Specified by:

createData in interface IDataService

createData

protected Data createData(ISqlTransaction transaction,
 Trigger trigger,
 String whereClause)

getCsvDataFor

protected String getCsvDataFor(ISqlTransaction transaction,
 Trigger trigger,
 TriggerHistory triggerHistory,
 String whereClause,
 boolean pkOnly)

countDataGaps

public long countDataGaps()

Specified by:

countDataGaps in interface IDataService

findDataGapsUnchecked

public List<DataGap> findDataGapsUnchecked()

Specified by:

findDataGapsUnchecked in interface IDataService

findDataGapsExpired

public List<DataGap> findDataGapsExpired()

Specified by:

findDataGapsExpired in interface IDataService

findDataGaps

protected List<DataGap> findDataGaps(boolean isExpired)

findDataGaps

public List<DataGap> findDataGaps()

Specified by:

findDataGaps in interface IDataService

findMaxDataEventDataId

public long findMaxDataEventDataId()

insertDataGap

public void insertDataGap(DataGap gap)

Specified by:

insertDataGap in interface IDataService

insertDataGap

public void insertDataGap(ISqlTransaction transaction,
 DataGap gap)

Specified by:

insertDataGap in interface IDataService

insertDataGaps

public void insertDataGaps(ISqlTransaction transaction,
 Collection<DataGap> gaps)

Specified by:

insertDataGaps in interface IDataService

deleteDataGap

public void deleteDataGap(DataGap gap)

Specified by:

deleteDataGap in interface IDataService

deleteDataGap

public void deleteDataGap(ISqlTransaction transaction,
 DataGap gap)

Specified by:

deleteDataGap in interface IDataService

deleteDataGaps

public void deleteDataGaps(ISqlTransaction transaction,
 Collection<DataGap> gaps)

Specified by:

deleteDataGaps in interface IDataService

deleteAllDataGaps

public void deleteAllDataGaps(ISqlTransaction transaction)

Specified by:

deleteAllDataGaps in interface IDataService

expireDataGaps

public void expireDataGaps(ISqlTransaction transaction,
 Collection<DataGap> gaps)

Specified by:

expireDataGaps in interface IDataService

findCreateTimeOfEvent

public Date findCreateTimeOfEvent(long dataId)

Specified by:

findCreateTimeOfEvent in interface IDataService

findCreateTimeOfData

public Date findCreateTimeOfData(long dataId)

Specified by:

findCreateTimeOfData in interface IDataService

findNextCreateTimeOfDataStartingAt

public Date findNextCreateTimeOfDataStartingAt(long dataId)

Specified by:

findNextCreateTimeOfDataStartingAt in interface IDataService

getHeartbeatListeners

protected List<IHeartbeatListener> getHeartbeatListeners(boolean force)

Get a list of IHeartbeatListeners that are ready for a heartbeat according to IHeartbeatListener.getTimeBetweenHeartbeatsInSeconds()

Parameters:

force - if true, then return the entire list of IHeartbeatListeners

updateLastHeartbeatTime

protected void updateLastHeartbeatTime(List<IHeartbeatListener> listeners)

heartbeat

public void heartbeat(boolean force)

Description copied from interface: IDataService

Update Node information for this node and call IHeartbeatListeners.

Specified by:

heartbeat in interface IDataService

See Also:

• IDataService#heartbeat()

listDataIds

public List<Number> listDataIds(long batchId,
 String nodeId)

Specified by:

listDataIds in interface IDataService

listData

public List<Data> listData(long batchId,
 String nodeId,
 long startDataId,
 String channelId,
 int maxRowsToRetrieve)

Specified by:

listData in interface IDataService

findData

public Data findData(long dataId)

Specified by:

findData in interface IDataService

findData

public List<Data> findData(long startDataId,
 long endDataId)

getDataMapper

public ISqlRowMapper<Data> getDataMapper()

Specified by:

getDataMapper in interface IDataService

selectDataFor

public ISqlReadCursor<Data> selectDataFor(Batch batch)

Specified by:

selectDataFor in interface IDataService

selectDataFor

public ISqlReadCursor<Data> selectDataFor(Long batchId,
 String targetNodeId,
 boolean isContainsBigLob)

Specified by:

selectDataFor in interface IDataService

selectDataFor

public ISqlReadCursor<Data> selectDataFor(Long batchId,
 String channelId)

Specified by:

selectDataFor in interface IDataService

getDataSelectByBatchSql

protected String getDataSelectByBatchSql(long batchId,
 long startDataId,
 String channelId)

getDataSelectSql

protected String getDataSelectSql(long batchId,
 long startDataId,
 String channelId)

getDataSelectSql

protected String getDataSelectSql(long batchId,
 long startDataId,
 boolean isContainsBigLob)

getDataOrderBy

protected String getDataOrderBy()

findMaxDataId

public long findMaxDataId()

Specified by:

findMaxDataId in interface IDataService

findMinDataId

public long findMinDataId()

Specified by:

findMinDataId in interface IDataService

deleteCapturedConfigChannelData

public void deleteCapturedConfigChannelData()

Specified by:

deleteCapturedConfigChannelData in interface IDataService

getLastDataCaptureByChannel

public Map<String,Date> getLastDataCaptureByChannel()

Specified by:

getLastDataCaptureByChannel in interface IDataService

fixLastDataGap

public boolean fixLastDataGap()

Specified by:

fixLastDataGap in interface IDataService

findNodeIdsByNodeGroupId

public String findNodeIdsByNodeGroupId()

Specified by:

findNodeIdsByNodeGroupId in interface IDataService

checkInterrupted

protected void checkInterrupted()
                         throws InterruptedException

Throws:

InterruptedException

resendBatchAsReload

public int resendBatchAsReload(long batchId,
 String nodeId)

Specified by:

resendBatchAsReload in interface IDataService

resendDataAsReload

public int resendDataAsReload(long minDataId,
 long maxDataId)

Specified by:

resendDataAsReload in interface IDataService

resendDataAsReload

protected void resendDataAsReload(List<Data> dataList,
 String nodeId)

convertDataToReload

protected void convertDataToReload(Data data,
 Table table,
 TriggerHistory hist,
 String nodeId)

reCaptureData

public int reCaptureData(long minDataId,
 long maxDataId)

Attempts to recapture stale data rows between specified Ids and re-insert them back into sym_data with a new transaction ID. Commits every RECAPTURE_DATA_COMMIT_LIMIT records to avoid locking sym_data from other processes.

Specified by:

reCaptureData in interface IDataService

Returns:

number of recaptured data rows

reCaptureData

protected int reCaptureData(List<Data> dataList)

Attempts to recapture stale data rows and re-insert them back into sym_data with a new transaction ID (all records are in one transaction).

Returns:

number of recaptured data rows

fetchRecapturedData

protected Data fetchRecapturedData(Data data,
 String recaptureTransactionId)

Attempts to recapture expired data row, which is still relevant and assigns a new transaction ID

Returns:

recaptured data row

insertRecapturedData

protected int insertRecapturedData(List<Data> insertList)

Inserts recaptured data back into sym_data

Returns:

number of inserted rows

recaptureKeysForData

protected String[] recaptureKeysForData(Table table,
 Data data)

Looks up values of primary keys for specified data row object

recaptureWhereFilterForKeys

protected String recaptureWhereFilterForKeys(Table table,
 TriggerHistory hist,
 String[] keys,
 Object[] values)

Extracts filter for a where clause for specified values of primary keys

hasColumnDataIntegrity

protected boolean hasColumnDataIntegrity(Data data,
 TriggerHistory hist)