You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "hackergin (via GitHub)" <gi...@apache.org> on 2024/06/05 03:44:20 UTC
[PR] [FLINK-35201][table] Support the execution of drop materialized table in full refresh mode [flink]
hackergin opened a new pull request, #24888:
URL: https://github.com/apache/flink/pull/24888
## What is the purpose of the change
*Support the execution of drop materialized table in full refresh mode*
## Brief change log
* Support the execution of drop materialized table in full refresh mode
## Verifying this change
* Add test case `testDropMaterializedTableInFullMode` to verify the execution of drop materialized table in full mode.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (Will be added in a separate pr)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35201][table] Support the execution of drop materialized table in full refresh mode [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on code in PR #24888:
URL: https://github.com/apache/flink/pull/24888#discussion_r1627675280
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java:
##########
@@ -782,35 +783,12 @@ private ResultFetcher callDropMaterializedTableOperation(
if (CatalogMaterializedTable.RefreshStatus.ACTIVATED
== materializedTable.getRefreshStatus()) {
- ContinuousRefreshHandler refreshHandler =
- deserializeContinuousHandler(materializedTable.getSerializedRefreshHandler());
- // get job running status
- JobStatus jobStatus = getJobStatus(operationExecutor, handle, refreshHandler);
- if (!jobStatus.isTerminalState()) {
- try {
- cancelJob(operationExecutor, handle, refreshHandler.getJobId());
- } catch (Exception e) {
- jobStatus = getJobStatus(operationExecutor, handle, refreshHandler);
- if (!jobStatus.isTerminalState()) {
- throw new SqlExecutionException(
- String.format(
- "Failed to drop the materialized table %s because the continuous refresh job %s could not be canceled."
- + " The current status of the continuous refresh job is %s.",
- tableIdentifier, refreshHandler.getJobId(), jobStatus),
- e);
- } else {
- LOG.warn(
- "An exception occurred while canceling the continuous refresh job {} for materialized table {},"
- + " but since the job is in a terminal state, skip the cancel operation.",
- refreshHandler.getJobId(),
- tableIdentifier);
- }
- }
+ if (CatalogMaterializedTable.RefreshMode.CONTINUOUS
+ == materializedTable.getRefreshMode()) {
+ cancelContinuousRefreshJob(
+ operationExecutor, handle, tableIdentifier, materializedTable);
} else {
- LOG.info(
- "No need to cancel the continuous refresh job {} for materialized table {} as it is not currently running.",
- refreshHandler.getJobId(),
- tableIdentifier);
+ deleteRefreshWorkflow(tableIdentifier, materializedTable);
}
} else if (CatalogMaterializedTable.RefreshStatus.INITIALIZING
Review Comment:
If the refresh status is `SUSPENDED`, we also need to drop the refresh workflow before drop the materialized table.
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/scheduler/EmbeddedQuartzScheduler.java:
##########
@@ -271,11 +271,6 @@ public void deleteScheduleWorkflow(String workflowName, String workflowGroup)
JobKey jobKey = JobKey.jobKey(workflowName, workflowGroup);
lock.writeLock().lock();
try {
- String errorMsg =
- String.format(
- "Failed to delete a non-existent quartz schedule job: %s.", jobKey);
- checkJobExists(jobKey, errorMsg);
-
quartzScheduler.deleteJob(jobKey);
Review Comment:
Do you verify whether `quartzScheduler.deleteJob` throws an exception when deleting a not exist job?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35201][table] Support the execution of drop materialized table in full refresh mode [flink]
Posted by "hackergin (via GitHub)" <gi...@apache.org>.
hackergin commented on code in PR #24888:
URL: https://github.com/apache/flink/pull/24888#discussion_r1628739971
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/scheduler/EmbeddedQuartzScheduler.java:
##########
@@ -271,11 +271,6 @@ public void deleteScheduleWorkflow(String workflowName, String workflowGroup)
JobKey jobKey = JobKey.jobKey(workflowName, workflowGroup);
lock.writeLock().lock();
try {
- String errorMsg =
- String.format(
- "Failed to delete a non-existent quartz schedule job: %s.", jobKey);
- checkJobExists(jobKey, errorMsg);
-
quartzScheduler.deleteJob(jobKey);
Review Comment:
Yes, In the testDropMaterializedTableWithDeletedRefreshWorkflowInFullMode test, even if the workflow does not exist, `quartzScheduler.deleteJob(jobKey)` will return normally without error.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35201][table] Support the execution of drop materialized table in full refresh mode [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu merged PR #24888:
URL: https://github.com/apache/flink/pull/24888
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35201][table] Support the execution of drop materialized table in full refresh mode [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24888:
URL: https://github.com/apache/flink/pull/24888#issuecomment-2148801220
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f0ab631e7d9fea91593ec7550454fb926b265a60",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f0ab631e7d9fea91593ec7550454fb926b265a60",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f0ab631e7d9fea91593ec7550454fb926b265a60 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35201][table] Support the execution of drop materialized table in full refresh mode [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on code in PR #24888:
URL: https://github.com/apache/flink/pull/24888#discussion_r1628807845
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java:
##########
@@ -780,37 +752,18 @@ private ResultFetcher callDropMaterializedTableOperation(
CatalogMaterializedTable materializedTable =
getCatalogMaterializedTable(operationExecutor, tableIdentifier);
- if (CatalogMaterializedTable.RefreshStatus.ACTIVATED
- == materializedTable.getRefreshStatus()) {
- ContinuousRefreshHandler refreshHandler =
- deserializeContinuousHandler(materializedTable.getSerializedRefreshHandler());
- // get job running status
- JobStatus jobStatus = getJobStatus(operationExecutor, handle, refreshHandler);
- if (!jobStatus.isTerminalState()) {
- try {
- cancelJob(operationExecutor, handle, refreshHandler.getJobId());
- } catch (Exception e) {
- jobStatus = getJobStatus(operationExecutor, handle, refreshHandler);
- if (!jobStatus.isTerminalState()) {
- throw new SqlExecutionException(
- String.format(
- "Failed to drop the materialized table %s because the continuous refresh job %s could not be canceled."
- + " The current status of the continuous refresh job is %s.",
- tableIdentifier, refreshHandler.getJobId(), jobStatus),
- e);
- } else {
- LOG.warn(
- "An exception occurred while canceling the continuous refresh job {} for materialized table {},"
- + " but since the job is in a terminal state, skip the cancel operation.",
- refreshHandler.getJobId(),
- tableIdentifier);
- }
+ if (CatalogMaterializedTable.RefreshStatus.ACTIVATED == materializedTable.getRefreshStatus()
Review Comment:
We should minimize the nesting depth of `if else` in code snippet, so can optimize the code:
```
CatalogMaterializedTable.RefreshMode refreshMode = materializedTable.getRefreshMode();
CatalogMaterializedTable.RefreshStatus refreshStatus = materializedTable.getRefreshStatus();
if (CatalogMaterializedTable.RefreshStatus.ACTIVATED == refreshStatus
|| CatalogMaterializedTable.RefreshStatus.SUSPENDED == refreshStatus) {
if (CatalogMaterializedTable.RefreshMode.FULL == refreshMode) {
deleteRefreshWorkflow(tableIdentifier, materializedTable);
} else if (CatalogMaterializedTable.RefreshMode.CONTINUOUS == refreshMode
&& CatalogMaterializedTable.RefreshStatus.ACTIVATED == refreshStatus) {
cancelContinuousRefreshJob(
operationExecutor, handle, tableIdentifier, materializedTable);
}
} else if (CatalogMaterializedTable.RefreshStatus.INITIALIZING == refreshStatus) {
throw new ValidationException(
String.format(
"Current refresh status of materialized table %s is initializing, skip the drop operation.",
tableIdentifier.asSerializableString()));
}
```
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java:
##########
@@ -268,13 +268,10 @@ void testCreateMaterializedTableInFullMode() throws Exception {
.containsKey("table.catalog-store.file.path")
.doesNotContainKey(WORKFLOW_SCHEDULER_TYPE.key())
.doesNotContainKey(RESOURCES_DOWNLOAD_DIR.key());
-
- // delete the workflow
- embeddedWorkflowScheduler.deleteScheduleWorkflow(jobKey.getName(), jobKey.getGroup());
}
@Test
- void testCreateMaterializedTableFailed() throws Exception {
+ void testCreateMaterializedTableFailedInInContinuousMode() throws Exception {
Review Comment:
`throws Exception` is not needed in the method signature.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org