You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by ji...@apache.org on 2023/02/04 02:45:26 UTC
[shardingsphere] branch master updated: Refactor `ShowMigrationJobStatusResultSet` (#23981)
This is an automated email from the ASF dual-hosted git repository.
jianglongtao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new fd426f1307a Refactor `ShowMigrationJobStatusResultSet` (#23981)
fd426f1307a is described below
commit fd426f1307abeb2c1ea60738dbcdffe91332ab39
Author: Zichao <57...@users.noreply.github.com>
AuthorDate: Sat Feb 4 15:45:17 2023 +1300
Refactor `ShowMigrationJobStatusResultSet` (#23981)
* Replace `ShowMigrationJobStatusResultSet` with `ShowMigrationJobStatusExecutor`
* Replace `ShowMigrationJobStatusResultSet` with `ShowMigrationJobStatusExecutor`
* Replace `ShowMigrationJobStatusResultSet` with `ShowMigrationJobStatusExecutor`
---
...et.java => ShowMigrationJobStatusExecutor.java} | 54 ++++++----------------
....distsql.handler.ral.query.QueryableRALExecutor | 1 +
...here.distsql.handler.resultset.DistSQLResultSet | 1 -
.../query/ShowMigrationJobStatusExecutorTest.java | 45 ++++++++++++++++++
4 files changed, 61 insertions(+), 40 deletions(-)
diff --git a/kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusResultSet.java b/kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusExecutor.java
similarity index 61%
rename from kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusResultSet.java
rename to kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusExecutor.java
index 5718e9454e0..5b4810d383e 100644
--- a/kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusResultSet.java
+++ b/kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusExecutor.java
@@ -21,59 +21,45 @@ import org.apache.shardingsphere.data.pipeline.api.job.progress.InventoryIncreme
import org.apache.shardingsphere.data.pipeline.api.pojo.InventoryIncrementalJobItemInfo;
import org.apache.shardingsphere.data.pipeline.core.api.InventoryIncrementalJobAPI;
import org.apache.shardingsphere.data.pipeline.core.api.PipelineJobAPI;
-import org.apache.shardingsphere.distsql.handler.resultset.DatabaseDistSQLResultSet;
-import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
+import org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor;
+import org.apache.shardingsphere.infra.merge.result.impl.local.LocalDataQueryResultRow;
import org.apache.shardingsphere.infra.util.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.migration.distsql.statement.ShowMigrationStatusStatement;
-import org.apache.shardingsphere.sql.parser.sql.common.statement.SQLStatement;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Iterator;
-import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
- * Result set for show migration job status.
+ * Show migration job status executor.
*/
-public final class ShowMigrationJobStatusResultSet implements DatabaseDistSQLResultSet {
-
- private Iterator<Collection<Object>> data;
+public final class ShowMigrationJobStatusExecutor implements QueryableRALExecutor<ShowMigrationStatusStatement> {
@Override
- public void init(final ShardingSphereDatabase database, final SQLStatement sqlStatement) {
+ public Collection<LocalDataQueryResultRow> getRows(final ShowMigrationStatusStatement sqlStatement) {
InventoryIncrementalJobAPI jobAPI = (InventoryIncrementalJobAPI) TypedSPILoader.getService(PipelineJobAPI.class, "MIGRATION");
- List<InventoryIncrementalJobItemInfo> jobItemInfos = jobAPI.getJobItemInfos(((ShowMigrationStatusStatement) sqlStatement).getJobId());
+ List<InventoryIncrementalJobItemInfo> jobItemInfos = jobAPI.getJobItemInfos(sqlStatement.getJobId());
long currentTimeMillis = System.currentTimeMillis();
- data = jobItemInfos.stream().map(each -> {
- Collection<Object> result = new LinkedList<>();
- result.add(each.getShardingItem());
+ Collection<LocalDataQueryResultRow> result = jobItemInfos.stream().map(each -> {
+ LocalDataQueryResultRow row;
InventoryIncrementalJobItemProgress jobItemProgress = each.getJobItemProgress();
if (null != jobItemProgress) {
- result.add(jobItemProgress.getDataSourceName());
- result.add(jobItemProgress.getStatus());
- result.add(jobItemProgress.isActive() ? Boolean.TRUE.toString() : Boolean.FALSE.toString());
- result.add(jobItemProgress.getProcessedRecordsCount());
- result.add(each.getInventoryFinishedPercentage());
String incrementalIdleSeconds = "";
if (jobItemProgress.getIncremental().getIncrementalLatestActiveTimeMillis() > 0) {
long latestActiveTimeMillis = Math.max(each.getStartTimeMillis(), jobItemProgress.getIncremental().getIncrementalLatestActiveTimeMillis());
incrementalIdleSeconds = String.valueOf(TimeUnit.MILLISECONDS.toSeconds(currentTimeMillis - latestActiveTimeMillis));
}
- result.add(incrementalIdleSeconds);
+ row = new LocalDataQueryResultRow(each.getShardingItem(), jobItemProgress.getDataSourceName(), jobItemProgress.getStatus(),
+ jobItemProgress.isActive() ? Boolean.TRUE.toString() : Boolean.FALSE.toString(), jobItemProgress.getProcessedRecordsCount(), each.getInventoryFinishedPercentage(),
+ incrementalIdleSeconds, each.getErrorMessage());
} else {
- result.add("");
- result.add("");
- result.add("");
- result.add("");
- result.add("");
- result.add("");
+ row = new LocalDataQueryResultRow(each.getShardingItem(), "", "", "", "", "", "", each.getErrorMessage());
}
- result.add(each.getErrorMessage());
- return result;
- }).collect(Collectors.toList()).iterator();
+ return row;
+ }).collect(Collectors.toList());
+ return result;
}
@Override
@@ -81,16 +67,6 @@ public final class ShowMigrationJobStatusResultSet implements DatabaseDistSQLRes
return Arrays.asList("item", "data_source", "status", "active", "processed_records_count", "inventory_finished_percentage", "incremental_idle_seconds", "error_message");
}
- @Override
- public boolean next() {
- return data.hasNext();
- }
-
- @Override
- public Collection<Object> getRowData() {
- return data.next();
- }
-
@Override
public String getType() {
return ShowMigrationStatusStatement.class.getName();
diff --git a/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor b/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor
index ad4abdc5252..9370d7d3f7b 100644
--- a/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor
+++ b/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor
@@ -16,3 +16,4 @@
#
org.apache.shardingsphere.migration.distsql.handler.query.ShowMigrationListExecutor
+org.apache.shardingsphere.migration.distsql.handler.query.ShowMigrationJobStatusExecutor
diff --git a/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.resultset.DistSQLResultSet b/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.resultset.DistSQLResultSet
index 30f68c6ace0..1054006ae9d 100644
--- a/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.resultset.DistSQLResultSet
+++ b/kernel/data-pipeline/distsql/handler/src/main/resources/META-INF/services/org.apache.shardingsphere.distsql.handler.resultset.DistSQLResultSet
@@ -16,6 +16,5 @@
#
org.apache.shardingsphere.migration.distsql.handler.query.ShowMigrationCheckStatusResultSet
-org.apache.shardingsphere.migration.distsql.handler.query.ShowMigrationJobStatusResultSet
org.apache.shardingsphere.migration.distsql.handler.query.ShowMigrationCheckAlgorithmsResultSet
org.apache.shardingsphere.migration.distsql.handler.query.ShowMigrationSourceStorageUnitsResultSet
diff --git a/kernel/data-pipeline/distsql/handler/src/test/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusExecutorTest.java b/kernel/data-pipeline/distsql/handler/src/test/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusExecutorTest.java
new file mode 100644
index 00000000000..ad4200b6078
--- /dev/null
+++ b/kernel/data-pipeline/distsql/handler/src/test/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationJobStatusExecutorTest.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.migration.distsql.handler.query;
+
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public final class ShowMigrationJobStatusExecutorTest {
+
+ @Test
+ public void assertGetColumnNames() {
+ ShowMigrationJobStatusExecutor executor = new ShowMigrationJobStatusExecutor();
+ Collection<String> columns = executor.getColumnNames();
+ assertThat(columns.size(), is(8));
+ Iterator<String> iterator = columns.iterator();
+ assertThat(iterator.next(), is("item"));
+ assertThat(iterator.next(), is("data_source"));
+ assertThat(iterator.next(), is("status"));
+ assertThat(iterator.next(), is("active"));
+ assertThat(iterator.next(), is("processed_records_count"));
+ assertThat(iterator.next(), is("inventory_finished_percentage"));
+ assertThat(iterator.next(), is("incremental_idle_seconds"));
+ assertThat(iterator.next(), is("error_message"));
+ }
+}