Skip to content

[hotfix] Fix grammatical errors in Materialized Table and related code #28050

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rmoff wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[hotfix] Fix grammatical errors in Materialized Table and related code #28050

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions ...va/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,13 +232,13 @@ private void createMaterializedTableInContinuousMode(
Collections.emptyMap(),
Optional.empty());
} catch (Exception e) {
// drop materialized table while submit flink streaming job occur exception. Thus, weak
// drop materialized table if submitting the Flink streaming job encounters an exception. Thus, weak
// atomicity is guaranteed
operationExecutor.callExecutableOperation(
handle, new DropMaterializedTableOperation(materializedTableIdentifier, true));
throw new SqlExecutionException(
String.format(
"Submit continuous refresh job for materialized table %s occur exception.",
"Failed to submit continuous refresh job for materialized table %s.",
materializedTableIdentifier),
e);
}
Expand Down Expand Up @@ -288,7 +288,7 @@ private void createMaterializedTableInFullMode(
refreshHandler.asSummaryString(),
serializedRefreshHandler);
} catch (Exception e) {
// drop materialized table while create refresh workflow occur exception. Thus, weak
// drop materialized table if creating the refresh workflow encounters an exception. Thus, weak
// atomicity is guaranteed
operationExecutor.callExecutableOperation(
handle, new DropMaterializedTableOperation(materializedTableIdentifier, true));
Expand Down Expand Up @@ -651,7 +651,7 @@ public ResultFetcher refreshMaterializedTable(

try {
LOG.info(
"Begin to refreshing the materialized table {}, statement: {}",
"Starting refresh of the materialized table {}, statement: {}",
materializedTableIdentifier,
insertStatement);
JobExecutionResult result =
Expand Down Expand Up @@ -682,7 +682,7 @@ public ResultFetcher refreshMaterializedTable(
} catch (Exception e) {
throw new SqlExecutionException(
String.format(
"Refreshing the materialized table %s occur exception.",
"Failed to refresh the materialized table %s.",
materializedTableIdentifier),
e);
}
Expand Down Expand Up @@ -1129,7 +1129,7 @@ private ContinuousRefreshHandler deserializeContinuousHandler(byte[] serializedR
serializedRefreshHandler, userCodeClassLoader);
} catch (IOException | ClassNotFoundException e) {
throw new SqlExecutionException(
"Deserialize ContinuousRefreshHandler occur exception.", e);
"Failed to deserialize ContinuousRefreshHandler.", e);
}
}

Expand All @@ -1138,7 +1138,7 @@ private byte[] serializeContinuousHandler(ContinuousRefreshHandler refreshHandle
return ContinuousRefreshHandlerSerializer.INSTANCE.serialize(refreshHandler);
} catch (IOException e) {
throw new SqlExecutionException(
"Serialize ContinuousRefreshHandler occur exception.", e);
"Failed to serialize ContinuousRefreshHandler.", e);
}
}

Expand Down
2 changes: 1 addition & 1 deletion ...rc/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -492,7 +492,7 @@ void testCreateMaterializedTableFailedInInContinuousMode() {
.cause()
.hasMessageContaining(
String.format(
"Submit continuous refresh job for materialized table %s occur exception.",
"Failed to submit continuous refresh job for materialized table %s.",
userShopsIdentifier.asSerializableString()));

// verify the materialized table is not created
Expand Down
2 changes: 1 addition & 1 deletion ...e/flink-table-common/src/main/java/org/apache/flink/table/workflow/WorkflowScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ public interface WorkflowScheduler<T extends RefreshHandler> {
* Open this workflow scheduler instance. Used for any required preparation in initialization
* phase.
*
* @throws WorkflowException if initializing workflow scheduler occur exception
* @throws WorkflowException if initializing the workflow scheduler fails
*/
void open() throws WorkflowException;

Expand Down
27 changes: 11 additions & 16 deletions ...ls/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ public void open() throws CatalogException {
}
} catch (IOException e) {
throw new CatalogException(
String.format("Checking catalog path %s exists occur exception.", catalogPath),
String.format("Error checking whether catalog path %s exists.", catalogPath),
e);
}
}
Expand All @@ -146,7 +146,7 @@ public List<String> listDatabases() throws CatalogException {
.map(fileStatus -> fileStatus.getPath().getName())
.collect(Collectors.toList());
} catch (IOException e) {
throw new CatalogException("Listing database occur exception.", e);
throw new CatalogException("Error listing databases.", e);
}
}

Expand Down Expand Up @@ -189,8 +189,7 @@ public void createDatabase(String name, CatalogDatabase database, boolean ignore
try {
fs.mkdirs(dbPath);
} catch (IOException e) {
throw new CatalogException(
String.format("Creating database %s occur exception.", name), e);
throw new CatalogException(String.format("Error creating database %s.", name), e);
}
}

Expand Down Expand Up @@ -220,7 +219,7 @@ public void dropDatabase(String databaseName, boolean ignoreIfNotExists, boolean
fs.delete(dbPath, true);
} catch (IOException e) {
throw new CatalogException(
String.format("Dropping database %s occur exception.", databaseName), e);
String.format("Error dropping database %s.", databaseName), e);
}
}

Expand All @@ -246,7 +245,7 @@ public List<String> listTables(String databaseName)
.collect(Collectors.toList());
} catch (IOException e) {
throw new CatalogException(
String.format("Listing table in database %s occur exception.", dbPath), e);
String.format("Error listing tables in database %s.", dbPath), e);
}
}

Expand Down Expand Up @@ -276,8 +275,7 @@ public CatalogBaseTable getTable(ObjectPath tablePath)
tableInfo.getCatalogTableInfo(),
tableDataPath.toString());
} catch (IOException e) {
throw new CatalogException(
String.format("Getting table %s occur exception.", tablePath), e);
throw new CatalogException(String.format("Error getting table %s.", tablePath), e);
}
}

Expand All @@ -298,7 +296,7 @@ public boolean tableExists(ObjectPath tablePath) throws CatalogException {
return fs.exists(path) && fs.exists(tableSchemaFilePath);
} catch (IOException e) {
throw new CatalogException(
String.format("Checking table %s exists occur exception.", tablePath), e);
String.format("Error checking whether table %s exists.", tablePath), e);
}
}

Expand All @@ -317,8 +315,7 @@ public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
try {
fs.delete(path, true);
} catch (IOException e) {
throw new CatalogException(
String.format("Dropping table %s occur exception.", tablePath), e);
throw new CatalogException(String.format("Error dropping table %s.", tablePath), e);
}
}

Expand Down Expand Up @@ -374,8 +371,7 @@ public void createTable(
}

} catch (IOException e) {
throw new CatalogException(
String.format("Create table %s occur exception.", tablePath), e);
throw new CatalogException(String.format("Error creating table %s.", tablePath), e);
}
}

Expand Down Expand Up @@ -404,7 +400,7 @@ public void alterTable(
if (!fs.exists(tableSchemaPath)) {
throw new CatalogException(
String.format(
"Table %s schema file %s doesn't exists.",
"Table %s schema file %s doesn't exist.",
tablePath, tableSchemaPath));
}
// write new table schema
Expand All @@ -416,8 +412,7 @@ public void alterTable(
}

} catch (IOException e) {
throw new CatalogException(
String.format("Altering table %s occur exception.", tablePath), e);
throw new CatalogException(String.format("Error altering table %s.", tablePath), e);
}
}

Expand Down