github-actions[bot] commented on code in PR #69673:
URL: https://github.com/apache/doris/pull/69673#discussion_r4387450700
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -599,6 +621,48 @@ public void alterJob(AlterJobCommand alterJobCommand)
throws AnalysisException,
}
}
+ private void alterIncludeTables(AlterJobCommand command, List<String>
updatedTables,
+ Map<String, String> newConvertedSourceProperties, Map<String,
String> mergedTargetProperties)
+ throws AnalysisException, JobException {
+ if (updatedTables == null || new HashSet<>(updatedTables).equals(new
HashSet<>(syncTables))) {
+ return;
+ }
+ if
(command.getProperties().containsKey(StreamingJobProperties.OFFSET_PROPERTY)) {
+ throw new AnalysisException("Cannot alter include_tables and
offset together");
+ }
+ // A paused job may not have restored its committed progress after an
FE restart.
+ replayOffsetProviderIfNeed();
+ offsetProvider.validateAlterIncludeTables(updatedTables);
+ List<String> addedTables = new ArrayList<>(updatedTables);
+ addedTables.removeAll(syncTables);
+ prepareIncludeTablesChange(newConvertedSourceProperties,
mergedTargetProperties, updatedTables, addedTables);
+
+ offsetProvider.applyIncludeTablesChange(updatedTables, addedTables);
+ offsetProvider.ensureInitialized(getJobId(),
newConvertedSourceProperties);
+ syncTables = updatedTables;
+ offsetProviderPersist = offsetProvider.getPersistInfo();
+ needRebuildReader = true;
+ }
+
+ private void prepareIncludeTablesChange(Map<String, String>
newConvertedProperties,
+ Map<String, String> newTargetProperties, List<String>
updatedTables, List<String> addedTables)
+ throws JobException {
+ try {
+ StreamingJobUtils.validateAndCreateTables(targetDb, dataSourceType,
Review Comment:
[P1] Prepare added targets before taking the JobManager lock
JobManager.alterJob() holds its global write lock throughout this call. If
an added table has no Doris target, validateAndCreateTables() runs
CreateTableCommand.run() and eventually acquires the target database write lock
in InternalCatalog.createOlapTable(). Concurrent DROP MATERIALIZED VIEW in that
database holds the database write lock and calls unprotectDropTable() ->
MTMVService.dropJob() -> MTMVJobManager.dropJob() ->
JobManager.dropJobInternal(), which waits for the global manager lock. Thus
ALTER can own the manager lock while waiting for the database, and DROP can own
the database while waiting for the manager. Both acquisitions are unbounded,
blocking database DDL and global job operations indefinitely. Prepare the
targets outside the global manager lock, then reacquire locks in the documented
database-before-manager order and revalidate the paused job/table-set state
before publishing the change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
