Repository: apache/skywalking (language: python)

Issue:
[Feature] [Testing] Add IT to cover ShardingSphere DistSQL usage
### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/skywalking/issues?q=is%3Aissue) and found no similar feature requirement.


### Description

We are using DistSQL in Sharding-MySQL storage option, but DistSQL is a new tech stack, and not stable.

The stability concern is noticed from 5.1.2 to 5.2.x version bumping. We have successfully passed all tests in 5.1.2 but failed in all 5.2 dev and released versions.

So, in order to ease the ShardingSphere community verification, we could build an Integration Testing based on test container similar to ElasticSearch client test.

### Use case

_No response_

### Related issues

_No response_

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)



Discussion / hints:

Fix patch:
diff --git a/.github/workflows/skywalking.yaml b/.github/workflows/skywalking.yaml
index 786640d32c5a..621883f20926 100644
--- a/.github/workflows/skywalking.yaml
+++ b/.github/workflows/skywalking.yaml
@@ -271,6 +271,36 @@ jobs:
           ./mvnw clean verify -q -B -Dcheckstyle.skip -DskipUTs -DskipITs=false || \
           ./mvnw clean verify -q -B -Dcheckstyle.skip -DskipUTs -DskipITs=false
 
+  it-sharding-proxy:
+    if: |
+      always() &&
+      ((github.event_name == 'schedule' && github.repository == 'apache/skywalking') || needs.changes.outputs.oap == 'true')
+    name: IT Sharding-proxy
+    needs: [sanity-check, changes]
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    strategy:
+      matrix:
+        java-version: [8, 11]
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          submodules: true
+      - name: Cache maven repository
+        uses: actions/cache@v3
+        with:
+          path: ~/.m2/repository
+          key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
+          restore-keys: ${{ runner.os }}-maven-
+      - uses: actions/setup-java@v3
+        with:
+          java-version: ${{ matrix.java-version }}
+          distribution: adopt
+      - name: IT Sharding-proxy
+        run: |
+          ./mvnw clean  test -Dtest=org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.ShardingIntegrationTest -DfailIfNoTests=false || \
+          ./mvnw clean  test -Dtest=org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.ShardingIntegrationTest -DfailIfNoTests=false
+
   e2e-test:
     if: |
       always() &&
@@ -715,6 +745,7 @@ jobs:
       - dependency-license
       - unit-test
       - integration-test
+      - it-sharding-proxy
       - e2e-test
       - e2e-test-istio
       - e2e-test-java-versions
@@ -733,6 +764,7 @@ jobs:
           depLicenseResults=${{ needs.dependency-license.result }}
           unitResults=${{ needs.unit-test.result }};
           integrationResults=${{ needs.integration-test.result }};
+          itShardingResults=${{ needs.it-sharding-proxy.result }};
           e2eResults=${{ needs.e2e-test.result }};
           e2eIstioResults=${{ needs.e2e-test-istio.result }};
           e2eJavaVersionResults=${{ needs.e2e-test-java-versions.result }};
@@ -743,5 +775,6 @@ jobs:
           [[ ${e2eResults} == 'success' ]] || [[ ${execute} != 'true' && ${e2eResults} == 'skipped' ]] || exit -5;
           [[ ${e2eIstioResults} == 'success' ]] || [[ ${execute} != 'true' && ${e2eIstioResults} == 'skipped' ]] || exit -6;
           [[ ${e2eJavaVersionResults} == 'success' ]] || [[ ${execute} != 'true' && ${e2eJavaVersionResults} == 'skipped' ]] || exit -7;
+          [[ ${itShardingResults} == 'success' ]] || [[ ${execute} != 'true' && ${itShardingResults} == 'skipped' ]] || exit -8;
 
           exit 0;
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 4a9457e0ba3e..b52cfada6a69 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -112,6 +112,7 @@
 * Add `LongText` to support longer logs persistent as a text type in ElasticSearch, instead of a keyword, to avoid length limitation.
 * Fix wrong system variable name `SW_CORE_ENABLE_ENDPOINT_NAME_GROUPING_BY_OPENAPI`. It was **opaenapi**.
 * Fix not-time-series model blocking OAP boots in no-init mode.
+* Fix `ShardingTopologyQueryDAO.loadServiceRelationsDetectedAtServerSide` invoke backend miss parameter `serviceIds`.
 * Changed system variable `SW_SUPERDATASET_STORAGE_DAY_STEP` to `SW_STORAGE_ES_SUPER_DATASET_DAY_STEP` to be consistent with other ES storage related variables.
 * Fix ESEventQueryDAO missing metric_table boolQuery criteria.
 
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/pom.xml b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/pom.xml
index a1b614020dc7..1c204849cc89 100644
--- a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/pom.xml
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/pom.xml
@@ -45,5 +45,11 @@
             <artifactId>storage-jdbc-hikaricp-plugin</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>8.0.13</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/dao/ShardingTopologyQueryDAO.java b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/dao/ShardingTopologyQueryDAO.java
index cf81ca06ce53..40eb704e21b3 100644
--- a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/dao/ShardingTopologyQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/dao/ShardingTopologyQueryDAO.java
@@ -35,7 +35,7 @@ public ShardingTopologyQueryDAO(JDBCHikariCPClient h2Client) {
     @Override
     public List<Call.CallDetail> loadServiceRelationsDetectedAtServerSide(Duration duration,
                                                                           List<String> serviceIds) throws IOException {
-        return super.loadServiceRelationsDetectedAtServerSide(DurationWithinTTL.INSTANCE.getMetricDurationWithinTTL(duration));
+        return super.loadServiceRelationsDetectedAtServerSide(DurationWithinTTL.INSTANCE.getMetricDurationWithinTTL(duration), serviceIds);
     }
 
     @Override
diff --git a/pom.xml b/pom.xml
index aaf2228e8689..ab76862160b0 100755
--- a/pom.xml
+++ b/pom.xml
@@ -339,6 +339,7 @@
                         <skip>${skipUTs}</skip>
                         <excludes>
                             <exclude>IT*.class</exclude>
+                            <exclude>ShardingIntegrationTest.class</exclude>
                         </excludes>
                     </configuration>
                 </plugin>


Test patch:
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/mysql/DurationWithinTTLTest.java b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/DurationWithinTTLTest.java
similarity index 98%
rename from oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/mysql/DurationWithinTTLTest.java
rename to oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/DurationWithinTTLTest.java
index 92598b62c535..f45f871db7ce 100644
--- a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/mysql/DurationWithinTTLTest.java
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/DurationWithinTTLTest.java
@@ -16,11 +16,10 @@
  *
  */
 
-package org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.mysql;
+package org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere;
 
 import org.apache.skywalking.oap.server.core.config.ConfigService;
 import org.apache.skywalking.oap.server.core.query.enumeration.Step;
-import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.DurationWithinTTL;
 import org.joda.time.DateTime;
 import org.junit.Assert;
 import org.junit.Before;
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ServiceCpmMetrics.java b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ServiceCpmMetrics.java
new file mode 100644
index 000000000000..2982b5186be7
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ServiceCpmMetrics.java
@@ -0,0 +1,102 @@
+/*
+ * 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.skywalking.oap.server.storage.plugin.jdbc.shardingsphere;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.skywalking.oap.server.core.Const;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.metrics.CPMMetrics;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
+import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+
+@Stream(
+    name = ServiceCpmMetrics.INDEX_NAME,
+    scopeId = DefaultScopeDefine.SERVICE,
+    builder = ServiceCpmMetricsBuilder.class,
+    processor = MetricsStreamProcessor.class
+)
+@EqualsAndHashCode(of = {
+    "entityId"
+}, callSuper = true)
+public class ServiceCpmMetrics extends CPMMetrics {
+    public static final String INDEX_NAME = "service_cpm";
+
+    @Setter
+    @Getter
+    @Column(
+        columnName = "entity_id",
+        length = 512
+    )
+    private String entityId;
+
+    @Override
+    protected String id0() {
+        return getTimeBucket() + Const.ID_CONNECTOR + entityId;
+    }
+
+    @Override
+    public int remoteHashCode() {
+        byte var1 = 17;
+        int var2 = 31 * var1 + this.entityId.hashCode();
+        return var2;
+    }
+
+    @Override
+    public RemoteData.Builder serialize() {
+        RemoteData.Builder var1 = RemoteData.newBuilder();
+        var1.addDataStrings(this.getEntityId());
+        var1.addDataLongs(this.getValue());
+        var1.addDataLongs(this.getTotal());
+        var1.addDataLongs(this.getTimeBucket());
+        return var1;
+    }
+
+    @Override
+    public void deserialize(RemoteData var1) {
+        this.setEntityId(var1.getDataStrings(0));
+        this.setValue(var1.getDataLongs(0));
+        this.setTotal(var1.getDataLongs(1));
+        this.setTimeBucket(var1.getDataLongs(2));
+    }
+
+    @Override
+    public Metrics toHour() {
+        ServiceCpmMetrics var1 = new ServiceCpmMetrics();
+        var1.setEntityId(this.getEntityId());
+        var1.setValue(this.getValue());
+        var1.setTotal(this.getTotal());
+        var1.setTimeBucket(this.toTimeBucketInHour());
+        return var1;
+    }
+
+    @Override
+    public Metrics toDay() {
+        ServiceCpmMetrics var1 = new ServiceCpmMetrics();
+        var1.setEntityId(this.getEntityId());
+        var1.setValue(this.getValue());
+        var1.setTotal(this.getTotal());
+        var1.setTimeBucket(this.toTimeBucketInDay());
+        return var1;
+    }
+}
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ServiceCpmMetricsBuilder.java b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ServiceCpmMetricsBuilder.java
new file mode 100644
index 000000000000..9269eb782205
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ServiceCpmMetricsBuilder.java
@@ -0,0 +1,46 @@
+/*
+ * 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.skywalking.oap.server.storage.plugin.jdbc.shardingsphere;
+
+import org.apache.skywalking.oap.server.core.storage.StorageData;
+import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity;
+import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage;
+import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
+
+public class ServiceCpmMetricsBuilder implements StorageBuilder {
+    public ServiceCpmMetricsBuilder() {
+    }
+
+    public void entity2Storage(StorageData var1, Convert2Storage var2) {
+        ServiceCpmMetrics var3 = (ServiceCpmMetrics) var1;
+        var2.accept("entity_id", var3.getEntityId());
+        var2.accept("value", new Long(var3.getValue()));
+        var2.accept("total", new Long(var3.getTotal()));
+        var2.accept("time_bucket", new Long(var3.getTimeBucket()));
+    }
+
+    public StorageData storage2Entity(Convert2Entity var1) {
+        ServiceCpmMetrics var2 = new ServiceCpmMetrics();
+        var2.setEntityId((String) var1.get("entity_id"));
+        var2.setValue(((Number) var1.get("value")).longValue());
+        var2.setTotal(((Number) var1.get("total")).longValue());
+        var2.setTimeBucket(((Number) var1.get("time_bucket")).longValue());
+        return var2;
+    }
+}
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ShardingIntegrationTest.java b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ShardingIntegrationTest.java
new file mode 100644
index 000000000000..c7ac50425666
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ShardingIntegrationTest.java
@@ -0,0 +1,732 @@
+/*
+ * 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.skywalking.oap.server.storage.plugin.jdbc.shardingsphere;
+
+import java.io.File;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.manual.endpoint.EndpointTraffic;
+import org.apache.skywalking.oap.server.core.analysis.manual.relation.service.ServiceRelationClientSideMetrics;
+import org.apache.skywalking.oap.server.core.analysis.manual.relation.service.ServiceRelationServerSideMetrics;
+import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.Tag;
+import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.TagAutocompleteData;
+import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.TagType;
+import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+import org.apache.skywalking.oap.server.core.analysis.record.Record;
+import org.apache.skywalking.oap.server.core.config.ConfigService;
+import org.apache.skywalking.oap.server.core.query.enumeration.Order;
+import org.apache.skywalking.oap.server.core.query.enumeration.Scope;
+import org.apache.skywalking.oap.server.core.query.enumeration.Step;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.core.query.input.Entity;
+import org.apache.skywalking.oap.server.core.query.input.MetricsCondition;
+import org.apache.skywalking.oap.server.core.query.input.TopNCondition;
+import org.apache.skywalking.oap.server.core.query.type.Call;
+import org.apache.skywalking.oap.server.core.query.type.Endpoint;
+import org.apache.skywalking.oap.server.core.query.type.MetricsValues;
+import org.apache.skywalking.oap.server.core.query.type.QueryOrder;
+import org.apache.skywalking.oap.server.core.query.type.SelectedRecord;
+import org.apache.skywalking.oap.server.core.query.type.TraceBrief;
+import org.apache.skywalking.oap.server.core.query.type.TraceState;
+import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
+import org.apache.skywalking.oap.server.core.storage.StorageData;
+import org.apache.skywalking.oap.server.core.storage.StorageException;
+import org.apache.skywalking.oap.server.core.storage.annotation.SQLDatabase;
+import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.core.storage.model.StorageModels;
+import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder;
+import org.apache.skywalking.oap.server.library.module.ModuleServiceHolder;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLExecutor;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.TableMetaInfo;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.common.dao.JDBCMetadataQueryDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.common.dao.JDBCMetricsDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.common.dao.JDBCRecordDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.common.dao.JDBCTagAutoCompleteQueryDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql.MySQLTableInstaller;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.dao.ShardingAggregationQueryDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.dao.ShardingHistoryDeleteDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.dao.ShardingMetricsQueryDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.dao.ShardingTopologyQueryDAO;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.dao.ShardingTraceQueryDAO;
+import org.joda.time.DateTime;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+import org.powermock.reflect.Whitebox;
+import org.testcontainers.containers.DockerComposeContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@Slf4j
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(Parameterized.class)
+@PowerMockIgnore({
+    "javax.net.*",
+    "javax.management.*",
+    "com.sun.org.apache.xerces.*",
+    "javax.xml.*", "org.xml.*",
+    "javax.management.*",
+    "org.w3c.*"
+})
+@PrepareForTest({DefaultScopeDefine.class})
+public class ShardingIntegrationTest {
+    @BeforeClass
+    public static void setup() {
+        PowerMockito.mockStatic(DefaultScopeDefine.class);
+        PowerMockito.when(DefaultScopeDefine.nameOf(1)).thenReturn("any");
+    }
+
+    @Parameterized.Parameter
+    public String version;
+
+    @Parameterized.Parameter(1)
+    public DataSourceType dsType;
+
+    @Parameterized.Parameters(name = "version: {0}")
+    public static Collection<Object[]> versions() {
+        return Arrays.asList(new Object[][] {
+            {
+                "5.1.2",
+                DataSourceType.MYSQL
+            }
+        });
+    }
+
+    public  DockerComposeContainer<?> environment;
+    private JDBCHikariCPClient ssClient;
+    private JDBCHikariCPClient dsClient0;
+    private JDBCHikariCPClient dsClient1;
+    private final Set<String> dataSources = new HashSet<>(Arrays.asList("ds_0", "ds_1"));
+    private final int ttlTestCreate = 3;
+    private final int ttlTestDrop = 2;
+    private final String searchableTag = "http.method";
+    private final long timeBucketSec = Long.parseLong(DateTime.now().toString("yyyyMMddHHmmss"));
+    private final long timeBucketMin = Long.parseLong(DateTime.now().toString("yyyyMMddHHmm"));
+    private final long timeBucketDay = Long.parseLong(DateTime.now().toString("yyyyMMdd"));
+    private Duration duration;
+    private Entity entityA;
+    private Entity entityB;
+    private String serviceIdA;
+    private String serviceIdB;
+
+    private ModuleManager moduleManager;
+    private MySQLTableInstaller mySQLTableInstaller;
+    private ShardingSphereTableInstaller installer;
+    private DurationWithinTTL durationWithinTTL = DurationWithinTTL.INSTANCE;
+    private final String countQuery = "SELECT COUNT(*) AS rc FROM ";
+
+    @Before
+    public void init() {
+        if (dsType.equals(DataSourceType.MYSQL)) {
+            startEnv("docker-compose-mysql.yml", 3306);
+            initConnection("mysql", "/swtest?rewriteBatchedStatements=true", 3306, "root", "root@1234");
+        }
+        initTestData();
+    }
+
+    private void startEnv(String dockerComposeName, int dsServicePort) {
+        environment = new DockerComposeContainer<>(new File(ShardingIntegrationTest.class
+                                                                .getClassLoader()
+                                                                .getResource(dockerComposeName).getPath()))
+            .withExposedService("sharding-proxy", 3307,
+                                Wait.defaultWaitStrategy().withStartupTimeout(java.time.Duration.ofMinutes(20))
+            )
+            .withExposedService("data-source-0", dsServicePort,
+                                Wait.defaultWaitStrategy().withStartupTimeout(java.time.Duration.ofMinutes(20))
+            )
+            .withExposedService("data-source-1", dsServicePort,
+                                Wait.defaultWaitStrategy().withStartupTimeout(java.time.Duration.ofMinutes(20))
+            )
+            .withEnv("SS_VERSION", version);
+        environment.start();
+    }
+
+    private void initConnection(String driverType,
+                                String urlSuffix,
+                                int dsServicePort,
+                                String dsUserName,
+                                String dsPassword) {
+        String ssUrl = "jdbc:" + driverType + "://" +
+            environment.getServiceHost("sharding-proxy", 3307) + ":" +
+            environment.getServicePort("sharding-proxy", 3307) +
+            urlSuffix;
+        Properties properties = new Properties();
+        properties.setProperty("jdbcUrl", ssUrl);
+        properties.setProperty("dataSource.user", "root");
+        properties.setProperty("dataSource.password", "root");
+
+        String dsUrl0 = "jdbc:" + driverType + "://" +
+            environment.getServiceHost("data-source-0", dsServicePort) + ":" +
+            environment.getServicePort("data-source-0", dsServicePort) +
+            urlSuffix;
+        Properties propertiesDs0 = new Properties();
+        propertiesDs0.setProperty("jdbcUrl", dsUrl0);
+        propertiesDs0.setProperty("dataSource.user", dsUserName);
+        propertiesDs0.setProperty("dataSource.password", "root@1234");
+
+        String dsUrl1 = "jdbc:" + driverType + "://" +
+            environment.getServiceHost("data-source-1", dsServicePort) + ":" +
+            environment.getServicePort("data-source-1", dsServicePort) +
+            urlSuffix;
+        Properties propertiesDs1 = new Properties();
+        propertiesDs1.setProperty("jdbcUrl", dsUrl1);
+        propertiesDs1.setProperty("dataSource.user", dsUserName);
+        propertiesDs1.setProperty("dataSource.password", dsPassword);
+
+        ssClient = new JDBCHikariCPClient(properties);
+        dsClient0 = new JDBCHikariCPClient(propertiesDs0);
+        dsClient1 = new JDBCHikariCPClient(propertiesDs1);
+
+        ssClient.connect();
+        dsClient0.connect();
+        dsClient1.connect();
+    }
+
+    private void initTestData() {
+        moduleManager = mock(ModuleManager.class);
+        ConfigService configService = mock(ConfigService.class);
+        Whitebox.setInternalState(moduleManager, "isInPrepareStage", false);
+        when(configService.getMetricsDataTTL()).thenReturn(ttlTestCreate);
+        when(configService.getRecordDataTTL()).thenReturn(ttlTestCreate);
+        when(configService.getSearchableTracesTags()).thenReturn(searchableTag);
+        when(moduleManager.find(anyString())).thenReturn(mock(ModuleProviderHolder.class));
+        when(moduleManager.find(CoreModule.NAME).provider()).thenReturn(mock(ModuleServiceHolder.class));
+        when(moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class))
+            .thenReturn(configService);
+        mySQLTableInstaller = new MySQLTableInstaller(ssClient, moduleManager);
+        installer = new ShardingSphereTableInstaller(ssClient, moduleManager,
+                                                     dataSources,
+                                                     mySQLTableInstaller
+        );
+
+        durationWithinTTL.setConfigService(configService);
+
+        duration = new Duration();
+        duration.setStart(DateTime.now().minusMinutes(15).toString("yyyy-MM-dd HHmm"));
+        duration.setEnd(DateTime.now().plusMinutes(1).toString("yyyy-MM-dd HHmm"));
+        duration.setStep(Step.MINUTE);
+
+        entityA = new Entity();
+        entityA.setServiceName("Service_A");
+        entityA.setNormal(true);
+        entityA.setScope(Scope.Service);
+
+        entityB = new Entity();
+        entityB.setServiceName("Service_B");
+        entityB.setNormal(true);
+        entityB.setScope(Scope.Service);
+
+        serviceIdA = entityA.buildId();
+        serviceIdB = entityB.buildId();
+    }
+
+    @SneakyThrows
+    @After
+    public void after() {
+        environment.stop();
+    }
+
+    @SneakyThrows
+    @Test
+    public void test() {
+        trafficTest();
+        metricsTest();
+        tagsTest();
+        recordsTest();
+        topologyTest();
+    }
+
+    // @SQLDatabase.Sharding(
+    // shardingAlgorithm = ShardingAlgorithm.TIME_MIN_RANGE_SHARDING_ALGORITHM,
+    // tableShardingColumn = TIME_BUCKET,
+    // dataSourceShardingColumn = ID)
+    @SneakyThrows
+    private void trafficTest() {
+        log.info("Traffic test start...");
+        StorageModels models = new StorageModels();
+        models.add(
+            EndpointTraffic.class, DefaultScopeDefine.SERVICE,
+            new Storage(EndpointTraffic.INDEX_NAME, false, DownSampling.Minute), false
+        );
+        Model model = models.allModels().get(0);
+        TableMetaInfo.addModel(model);
+
+        EndpointTraffic endpointTrafficA = new EndpointTraffic();
+        endpointTrafficA.setName("Endpoint_A");
+        endpointTrafficA.setServiceId(serviceIdA);
+        endpointTrafficA.setTimeBucket(timeBucketMin);
+        EndpointTraffic endpointTrafficB = new EndpointTraffic();
+        endpointTrafficB.setName("Endpoint_B");
+        endpointTrafficB.setServiceId(serviceIdB);
+        endpointTrafficB.setTimeBucket(timeBucketMin);
+        List<Metrics> metricsList = new ArrayList<>();
+        metricsList.add(endpointTrafficA);
+        metricsList.add(endpointTrafficB);
+
+        createShardingRuleTest(model);
+        updateShardingRuleTest(model);
+        createShardingTableTest(model);
+
+        insertMetrics(model, metricsList);
+        testDataSharding(endpointTrafficA);
+        ttlDropTest(model);
+
+        //Test traffic query
+        JDBCMetadataQueryDAO metadataQueryDAO = new JDBCMetadataQueryDAO(ssClient, 100);
+        List<Endpoint> endpoints = metadataQueryDAO.findEndpoint("", endpointTrafficA.getServiceId(), 100);
+        Assert.assertEquals(endpointTrafficA.getName(), endpoints.get(0).getName());
+        log.info("Traffic test passed.");
+    }
+
+    // @SQLDatabase.Sharding(
+    // shardingAlgorithm = ShardingAlgorithm.TIME_RELATIVE_ID_SHARDING_ALGORITHM,
+    // tableShardingColumn = ID,
+    // dataSourceShardingColumn = ENTITY_ID)
+    @SneakyThrows
+    private void metricsTest() {
+        log.info("Metrics test start...");
+        StorageModels models = new StorageModels();
+        models.add(
+            ServiceCpmMetrics.class, DefaultScopeDefine.SERVICE,
+            new Storage(ServiceCpmMetrics.INDEX_NAME, true, DownSampling.Minute), false
+        );
+        Model model = models.allModels().get(0);
+        TableMetaInfo.addModel(model);
+
+        ServiceCpmMetrics serviceCpmMetricsA = new ServiceCpmMetrics();
+        serviceCpmMetricsA.setEntityId(serviceIdA);
+        serviceCpmMetricsA.setValue(100);
+        serviceCpmMetricsA.setTotal(100);
+        serviceCpmMetricsA.setTimeBucket(timeBucketMin);
+        ServiceCpmMetrics serviceCpmMetricsB = new ServiceCpmMetrics();
+        serviceCpmMetricsB.setEntityId(serviceIdB);
+        serviceCpmMetricsB.setValue(200);
+        serviceCpmMetricsB.setTotal(200);
+        serviceCpmMetricsB.setTimeBucket(timeBucketMin);
+
+        createShardingRuleTest(model);
+        updateShardingRuleTest(model);
+        createShardingTableTest(model);
+
+        insertMetrics(model, Arrays.asList(serviceCpmMetricsA, serviceCpmMetricsB));
+        testDataSharding(serviceCpmMetricsA);
+
+        ttlDropTest(model);
+
+        //Test topN
+        ShardingAggregationQueryDAO aggregationQueryDAO = new ShardingAggregationQueryDAO(ssClient);
+        TopNCondition topNCondition = new TopNCondition();
+        topNCondition.setName(ServiceCpmMetrics.INDEX_NAME);
+        topNCondition.setTopN(1);
+        topNCondition.setOrder(Order.DES);
+
+        SelectedRecord top1Record = aggregationQueryDAO.sortMetrics(topNCondition, "value", duration, null).get(0);
+        Assert.assertEquals(serviceCpmMetricsB.getEntityId(), top1Record.getId());
+        Assert.assertEquals("200.0000", top1Record.getValue());
+
+        //Test metrics query
+        ShardingMetricsQueryDAO metricsQueryDAO = new ShardingMetricsQueryDAO(ssClient);
+        MetricsCondition metricsCondition = new MetricsCondition();
+
+        metricsCondition.setName(ServiceCpmMetrics.INDEX_NAME);
+        metricsCondition.setEntity(entityA);
+
+        long value = metricsQueryDAO.readMetricsValue(metricsCondition, "value", duration);
+        Assert.assertEquals(serviceCpmMetricsA.getValue(), value);
+
+        MetricsValues values = metricsQueryDAO.readMetricsValues(metricsCondition, "value", duration);
+        String metricsId = serviceCpmMetricsA.getTimeBucket() + "_" + serviceCpmMetricsA.getEntityId();
+        Assert.assertEquals(serviceCpmMetricsA.getValue(), values.getValues().findValue(metricsId, 0));
+        log.info("Metrics test passed.");
+    }
+
+    // @SQLDatabase.Sharding(shardingAlgorithm = ShardingAlgorithm.NO_SHARDING)
+    @SneakyThrows
+    private void tagsTest() {
+        log.info("Tag auto complete data test start...");
+        StorageModels models = new StorageModels();
+        models.add(
+            TagAutocompleteData.class, DefaultScopeDefine.SEGMENT,
+            new Storage(TagAutocompleteData.INDEX_NAME, true, DownSampling.Minute), false
+        );
+        Model model = models.allModels().get(0);
+        TableMetaInfo.addModel(model);
+        installer.createTable(model);
+        TagAutocompleteData tagData1 = new TagAutocompleteData();
+        tagData1.setTagType(TagType.TRACE.name());
+        tagData1.setTagKey(searchableTag);
+        tagData1.setTagValue("GET");
+        tagData1.setTimeBucket(timeBucketMin);
+        TagAutocompleteData tagData2 = new TagAutocompleteData();
+        tagData2.setTagType(TagType.TRACE.name());
+        tagData2.setTagKey(searchableTag);
+        tagData2.setTagValue("POST");
+        //Should be deleted after TTL process
+        tagData2.setTimeBucket(timeBucketMin);
+        TagAutocompleteData tagData3 = new TagAutocompleteData();
+        tagData3.setTagType(TagType.TRACE.name());
+        tagData3.setTagKey(searchableTag);
+        tagData3.setTagValue("HEAD");
+        //Should be deleted after TTL process
+        tagData3.setTimeBucket(Long.parseLong(DateTime.now().minusDays(5).toString("yyyyMMddHHmm")));
+        insertMetrics(model, Arrays.asList(tagData1, tagData2, tagData3));
+
+        //Test total count
+        try (Connection ssConn = ssClient.getConnection()) {
+            ResultSet rs = ssClient.executeQuery(ssConn, countQuery + TagAutocompleteData.INDEX_NAME);
+            rs.next();
+            Assert.assertEquals(3, rs.getInt("rc"));
+        }
+
+        // Test query
+        JDBCTagAutoCompleteQueryDAO tagQueryDAO = new JDBCTagAutoCompleteQueryDAO(ssClient);
+        Set<String> tagKeys = tagQueryDAO.queryTagAutocompleteKeys(TagType.TRACE, 10, duration);
+        Assert.assertEquals(searchableTag, tagKeys.iterator().next());
+        Set<String> tagValues = tagQueryDAO.queryTagAutocompleteValues(TagType.TRACE, searchableTag, 10, duration);
+        Assert.assertEquals(2, tagValues.size());
+
+        // Test TTL
+        historyDelete(model);
+        try (Connection ssConn = ssClient.getConnection()) {
+            ResultSet rs = ssClient.executeQuery(ssConn, countQuery + TagAutocompleteData.INDEX_NAME);
+            rs.next();
+            Assert.assertEquals(2, rs.getInt("rc"));
+        }
+        log.info("Tag auto complete data test passed.");
+    }
+
+    // @SQLDatabase.Sharding(
+    // shardingAlgorithm = ShardingAlgorithm.TIME_SEC_RANGE_SHARDING_ALGORITHM,
+    // dataSourceShardingColumn = SERVICE_ID,
+    // tableShardingColumn = TIME_BUCKET)
+    @SneakyThrows
+    private void recordsTest() {
+        log.info("Records (Trace) test start...");
+        StorageModels models = new StorageModels();
+        models.add(
+            SegmentRecord.class, DefaultScopeDefine.SEGMENT,
+            new Storage(SegmentRecord.INDEX_NAME, false, DownSampling.Second), true
+        );
+        Model model = models.allModels().get(0);
+        TableMetaInfo.addModel(model);
+        Tag tag = new Tag(searchableTag, "GET");
+        List<String> tags = Collections.singletonList(tag.toString());
+        SegmentRecord segmentRecordA = new SegmentRecord();
+        segmentRecordA.setSegmentId("segmentA");
+        segmentRecordA.setTraceId("traceA");
+        segmentRecordA.setServiceId(serviceIdA);
+        segmentRecordA.setTimeBucket(timeBucketSec);
+        segmentRecordA.setTags(tags);
+        segmentRecordA.setStartTime(DateTime.now().getMillis());
+        segmentRecordA.setEndpointId(IDManager.EndpointID.buildId(segmentRecordA.getServiceId(), "Endpoint_A"));
+        SegmentRecord segmentRecordB = new SegmentRecord();
+        segmentRecordB.setSegmentId("segmentB");
+        segmentRecordB.setTraceId("traceA");
+        segmentRecordB.setServiceId(serviceIdB);
+        segmentRecordB.setTimeBucket(timeBucketSec);
+        segmentRecordB.setTags(tags);
+        segmentRecordB.setStartTime(DateTime.now().getMillis());
+        segmentRecordA.setEndpointId(IDManager.EndpointID.buildId(segmentRecordB.getServiceId(), "Endpoint_B"));
+
+        createShardingRuleTest(model);
+        updateShardingRuleTest(model);
+        createShardingTableTest(model);
+        insertRecords(model, Arrays.asList(segmentRecordA, segmentRecordB));
+        testDataSharding(segmentRecordA);
+        ttlDropTest(model);
+
+        //Test trace query
+        ShardingTraceQueryDAO traceQueryDAO = new ShardingTraceQueryDAO(moduleManager, ssClient);
+        TraceBrief traceBrief = traceQueryDAO.queryBasicTraces(
+            duration, 0, 0,
+            segmentRecordA.getServiceId(), null,
+            null, null, 10, 0,
+            TraceState.SUCCESS, QueryOrder.BY_START_TIME, Collections.singletonList(tag)
+        );
+        Assert.assertEquals(segmentRecordA.getSegmentId(), traceBrief.getTraces().get(0).getSegmentId());
+
+        List<SegmentRecord> segmentRecords = traceQueryDAO.queryByTraceId(segmentRecordA.getTraceId());
+        Assert.assertEquals(2, segmentRecords.size());
+        log.info("Records (Trace) test passed.");
+    }
+
+    // @SQLDatabase.Sharding(
+    // shardingAlgorithm = ShardingAlgorithm.TIME_BUCKET_SHARDING_ALGORITHM,
+    // tableShardingColumn = TIME_BUCKET,
+    // dataSourceShardingColumn = ENTITY_ID)
+    @SneakyThrows
+    private void topologyTest() {
+        log.info("Topology test start...");
+        StorageModels models = new StorageModels();
+        models.add(
+            ServiceRelationServerSideMetrics.class, DefaultScopeDefine.SERVICE_RELATION,
+            new Storage(ServiceRelationServerSideMetrics.INDEX_NAME, true, DownSampling.Minute), false
+        );
+        models.add(
+            ServiceRelationClientSideMetrics.class, DefaultScopeDefine.SERVICE_RELATION,
+            new Storage(ServiceRelationClientSideMetrics.INDEX_NAME, true, DownSampling.Minute), false
+        );
+        Model serverModel = models.allModels().get(0);
+        Model clientModel = models.allModels().get(1);
+        TableMetaInfo.addModel(serverModel);
+        TableMetaInfo.addModel(clientModel);
+
+        ServiceRelationServerSideMetrics serverSideMetricsA = new ServiceRelationServerSideMetrics();
+        String clientIdA = IDManager.ServiceID.buildId("HTTP_Client", false);
+        serverSideMetricsA.setSourceServiceId(clientIdA);
+        serverSideMetricsA.setDestServiceId(serviceIdA);
+        serverSideMetricsA.setEntityId(IDManager.ServiceID.buildRelationId(
+            new IDManager.ServiceID.ServiceRelationDefine(clientIdA, serviceIdA)));
+        serverSideMetricsA.setComponentId(0);
+        serverSideMetricsA.setTimeBucket(timeBucketMin);
+        ServiceRelationClientSideMetrics clientSideMetricsA = new ServiceRelationClientSideMetrics();
+        clientSideMetricsA.setSourceServiceId(clientIdA);
+        clientSideMetricsA.setDestServiceId(serviceIdA);
+        clientSideMetricsA.setEntityId(IDManager.ServiceID.buildRelationId(
+            new IDManager.ServiceID.ServiceRelationDefine(clientIdA, serviceIdA)));
+        clientSideMetricsA.setComponentId(0);
+        clientSideMetricsA.setTimeBucket(timeBucketMin);
+
+        ServiceRelationServerSideMetrics serverSideMetricsB = new ServiceRelationServerSideMetrics();
+        serverSideMetricsB.setSourceServiceId(serviceIdA);
+        serverSideMetricsB.setDestServiceId(serviceIdB);
+        serverSideMetricsB.setEntityId(IDManager.ServiceID.buildRelationId(
+            new IDManager.ServiceID.ServiceRelationDefine(serviceIdA, serviceIdB)));
+        serverSideMetricsB.setComponentId(0);
+        serverSideMetricsB.setTimeBucket(timeBucketMin);
+        ServiceRelationClientSideMetrics clientSideMetricsB = new ServiceRelationClientSideMetrics();
+        clientSideMetricsB.setSourceServiceId(clientIdA);
+        clientSideMetricsB.setDestServiceId(serviceIdA);
+        clientSideMetricsB.setEntityId(IDManager.ServiceID.buildRelationId(
+            new IDManager.ServiceID.ServiceRelationDefine(serviceIdA, serviceIdB)));
+        clientSideMetricsB.setComponentId(0);
+        clientSideMetricsB.setTimeBucket(timeBucketMin);
+
+        createShardingRuleTest(serverModel);
+        createShardingRuleTest(clientModel);
+        updateShardingRuleTest(serverModel);
+        updateShardingRuleTest(clientModel);
+        createShardingTableTest(serverModel);
+        createShardingTableTest(clientModel);
+
+        insertMetrics(serverModel, Arrays.asList(serverSideMetricsA, serverSideMetricsB));
+        insertMetrics(clientModel, Arrays.asList(clientSideMetricsA, clientSideMetricsB));
+        testDataSharding(serverSideMetricsA);
+        testDataSharding(clientSideMetricsA);
+        ttlDropTest(serverModel);
+        ttlDropTest(clientModel);
+
+        //Test topology query
+        ShardingTopologyQueryDAO queryDAO = new ShardingTopologyQueryDAO(ssClient);
+        List<Call.CallDetail> callDetailsServerSide = queryDAO.loadServiceRelationsDetectedAtServerSide(
+            duration, Arrays.asList(serviceIdB));
+        //Service_A -----> Service_B
+        Assert.assertEquals(serviceIdB, callDetailsServerSide.get(0).getTarget());
+        List<Call.CallDetail> callDetailsClientSide = queryDAO.loadServiceRelationDetectedAtClientSide(
+            duration, Arrays.asList(serviceIdA));
+        //HTTP_Client -----> Service_A -----> Service_B
+        Assert.assertEquals(2, callDetailsClientSide.size());
+        log.info("Topology test passed.");
+    }
+
+    @SneakyThrows
+    private void createShardingRuleTest(Model model) {
+        ShardingRulesOperator.INSTANCE.createOrUpdateShardingRule(ssClient, model, dataSources, ttlTestCreate);
+        Map<String, ShardingRule> shardingRules = Whitebox.getInternalState(
+            ShardingRulesOperator.INSTANCE, "modelShardingRules");
+        ShardingRule inputRule = shardingRules.get(model.getName()).toBuilder().build();
+        ShardingRule outPutRule = loadShardingRule(model);
+        outPutRule.setOperation("CREATE");
+        //The rules in the database erased all `"`
+        Assert.assertEquals(inputRule.toShardingRuleSQL().replaceAll("\"", ""), outPutRule.toShardingRuleSQL());
+    }
+
+    private void updateShardingRuleTest(Model model) throws StorageException {
+        ShardingRulesOperator.INSTANCE.createOrUpdateShardingRule(ssClient, model, dataSources, ttlTestCreate);
+        Map<String, ShardingRule> shardingRules = Whitebox.getInternalState(
+            ShardingRulesOperator.INSTANCE, "modelShardingRules");
+        ShardingRule inputRule = shardingRules.get(model.getName()).toBuilder().build();
+        ShardingRule outPutRule = loadShardingRule(model);
+        outPutRule.setOperation("ALTER");
+        //The rules in the database erased all `"`
+        Assert.assertEquals(inputRule.toShardingRuleSQL().replaceAll("\"", ""), outPutRule.toShardingRuleSQL());
+    }
+
+    /**
+     * Load sharding rule from the database, return a clone rule for comparing.
+     */
+    @SneakyThrows
+    private ShardingRule loadShardingRule(Model model) {
+        Whitebox.invokeMethod(ShardingRulesOperator.INSTANCE, "initShardingRules", ssClient);
+        Map<String, ShardingRule> shardingRules = Whitebox.getInternalState(
+            ShardingRulesOperator.INSTANCE, "modelShardingRules");
+        return shardingRules.get(model.getName()).toBuilder().build();
+    }
+
+    @SneakyThrows
+    private void createShardingTableTest(Model model) {
+        installer.createTable(model);
+        existsTest(dsClient0, model);
+        existsTest(dsClient1, model);
+    }
+
+    @SneakyThrows
+    private void existsTest(JDBCHikariCPClient dsClient, Model model) {
+        //TTL is 3 so just test 4 tables
+        List<String> tables = new ArrayList<>();
+        tables.add(model.getName() + "_" + DateTime.now().minusDays(2).toString("yyyyMMdd"));
+        tables.add(model.getName() + "_" + DateTime.now().minusDays(1).toString("yyyyMMdd"));
+        tables.add(model.getName() + "_" + DateTime.now().toString("yyyyMMdd"));
+        tables.add(model.getName() + "_" + DateTime.now().plusDays(1).toString("yyyyMMdd"));
+        try (Connection conn = dsClient.getConnection()) {
+            for (String name : tables) {
+                ResultSet rset = conn.getMetaData().getTables(conn.getCatalog(), null, name, null);
+                Assert.assertTrue(rset.next());
+            }
+        }
+    }
+
+    @SneakyThrows
+    private void insertMetrics(Model model, List<Metrics> metricsList) {
+        try (Connection conn = ssClient.getConnection()) {
+            for (Metrics metrics : metricsList) {
+                JDBCMetricsDAO jdbcMetricsDAO = new JDBCMetricsDAO(ssClient, metrics.getClass()
+                                                                                    .getAnnotation(Stream.class)
+                                                                                    .builder()
+                                                                                    .getDeclaredConstructor()
+                                                                                    .newInstance());
+                jdbcMetricsDAO.prepareBatchInsert(model, metrics).invoke(conn);
+            }
+        }
+    }
+
+    @SneakyThrows
+    private void insertRecords(Model model, List<Record> recordList) {
+        try (Connection conn = ssClient.getConnection()) {
+            for (Record record : recordList) {
+                JDBCRecordDAO jdbcRecordDAO = new JDBCRecordDAO(ssClient, record.getClass()
+                                                                                .getAnnotation(Stream.class)
+                                                                                .builder()
+                                                                                .getDeclaredConstructor()
+                                                                                .newInstance());
+                ((SQLExecutor) jdbcRecordDAO.prepareBatchInsert(model, record)).invoke(conn);
+            }
+        }
+    }
+
+    @SneakyThrows
+    private void testDataSharding(StorageData data) {
+        try (Connection ssConn = ssClient.getConnection();
+             Connection ds0Conn = dsClient0.getConnection();
+             Connection ds1Conn = dsClient1.getConnection()) {
+
+            String logicIndex = data.getClass().getAnnotation(Stream.class).name();
+            String physicalIndex = logicIndex + "_" + timeBucketDay;
+
+            ResultSet logicSet = ssClient.executeQuery(ssConn, countQuery + logicIndex);
+            logicSet.next();
+            Assert.assertEquals(2, logicSet.getInt("rc"));
+
+            ResultSet physicalSet0 = dsClient0.executeQuery(ds0Conn, countQuery + physicalIndex);
+            physicalSet0.next();
+            Assert.assertEquals(1, physicalSet0.getInt("rc"));
+
+            ResultSet physicalSet1 = dsClient1.executeQuery(ds1Conn, countQuery + physicalIndex);
+            physicalSet1.next();
+            Assert.assertEquals(1, physicalSet1.getInt("rc"));
+
+            if (data.getClass().isAnnotationPresent(SQLDatabase.ExtraColumn4AdditionalEntity.class)) {
+                String additionalLogicIndex = data.getClass().getAnnotation(Stream.class).name();
+                String additionalPhysicalIndex = logicIndex + "_" + timeBucketDay;
+
+                ResultSet additionalLogicSet = ssClient.executeQuery(
+                    ssConn, countQuery + additionalLogicIndex);
+                additionalLogicSet.next();
+                Assert.assertEquals(2, additionalLogicSet.getInt("rc"));
+
+                ResultSet additionalPhysicalSet0 = dsClient0.executeQuery(
+                    ds0Conn, countQuery + additionalPhysicalIndex);
+                additionalPhysicalSet0.next();
+                Assert.assertEquals(1, additionalPhysicalSet0.getInt("rc"));
+
+                ResultSet additionalPhysicalSet1 = dsClient1.executeQuery(
+                    ds1Conn, countQuery + additionalPhysicalIndex);
+                additionalPhysicalSet1.next();
+                Assert.assertEquals(1, additionalPhysicalSet1.getInt("rc"));
+            }
+        }
+    }
+
+    @SneakyThrows
+    private void ttlDropTest(Model model) {
+        historyDelete(model);
+
+        String droppedTable = model.getName() + "_" + DateTime.now().minusDays(2).toString("yyyyMMdd");
+        ResultSet rset0 = dsClient0.getConnection()
+                                   .getMetaData()
+                                   .getTables(dsClient0.getConnection().getCatalog(), null, droppedTable, null);
+        Assert.assertFalse(rset0.next());
+        ResultSet rset1 = dsClient1.getConnection()
+                                   .getMetaData()
+                                   .getTables(dsClient1.getConnection().getCatalog(), null, droppedTable, null);
+        Assert.assertFalse(rset1.next());
+    }
+
+    @SneakyThrows
+    private void historyDelete(Model model) {
+        ShardingHistoryDeleteDAO deleteDAO = new ShardingHistoryDeleteDAO(
+            ssClient, dataSources, moduleManager, mySQLTableInstaller);
+        deleteDAO.deleteHistory(model, Metrics.TIME_BUCKET, ttlTestDrop);
+    }
+
+    private enum DataSourceType {
+        MYSQL
+    }
+}
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/mysql/ShardingRulesTest.java b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ShardingRulesTest.java
similarity index 96%
rename from oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/mysql/ShardingRulesTest.java
rename to oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ShardingRulesTest.java
index 6bbafef0318f..999f6f6d9ac5 100644
--- a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/mysql/ShardingRulesTest.java
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/shardingsphere/ShardingRulesTest.java
@@ -16,13 +16,11 @@
  *
  */
 
-package org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.mysql;
+package org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere;
 
 import java.util.Arrays;
 import java.util.HashSet;
 import org.apache.skywalking.oap.server.core.storage.ShardingAlgorithm;
-import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.ShardingRule;
-import org.apache.skywalking.oap.server.storage.plugin.jdbc.shardingsphere.ShardingRulesOperator;
 import org.joda.time.DateTime;
 import org.joda.time.format.DateTimeFormat;
 import org.junit.Assert;
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/conf-mysql/config-sharding.yaml b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/conf-mysql/config-sharding.yaml
new file mode 100644
index 000000000000..c7b336bd52fc
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/conf-mysql/config-sharding.yaml
@@ -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.
+#
+
+######################################################################################################
+# 
+# Here you can configure the rules for the proxy.
+# This example is configuration of sharding rule.
+# 
+######################################################################################################
+#
+databaseName: swtest
+
+dataSources:
+ ds_0:
+   url: jdbc:mysql://data-source-0:3306/swtest?allowPublicKeyRetrieval=true
+   username: root
+   password: root@1234
+   connectionTimeoutMilliseconds: 30000
+   idleTimeoutMilliseconds: 60000
+   maxLifetimeMilliseconds: 1800000
+   maxPoolSize: 50
+   minPoolSize: 1
+ ds_1:
+   url: jdbc:mysql://data-source-1:3306/swtest?allowPublicKeyRetrieval=true
+   username: root
+   password: root@1234
+   connectionTimeoutMilliseconds: 30000
+   idleTimeoutMilliseconds: 60000
+   maxLifetimeMilliseconds: 1800000
+   maxPoolSize: 50
+   minPoolSize: 1
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/conf-mysql/server.yaml b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/conf-mysql/server.yaml
new file mode 100644
index 000000000000..dc66c7cfe4ff
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/conf-mysql/server.yaml
@@ -0,0 +1,83 @@
+#
+# 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.
+#
+
+######################################################################################################
+#
+# If you want to configure governance, authorization and proxy properties, please refer to this file.
+#
+######################################################################################################
+
+#mode:
+#  type: Cluster
+#  repository:
+#    type: ZooKeeper
+#    props:
+#      namespace: governance_ds
+#      server-lists: localhost:2181
+#      retryIntervalMilliseconds: 500
+#      timeToLiveSeconds: 60
+#      maxRetries: 3
+#      operationTimeoutMilliseconds: 500
+#  overwrite: false
+#
+rules:
+ - !AUTHORITY
+   users:
+     - root@:root
+   provider:
+     type: ALL_PERMITTED
+ - !TRANSACTION
+   defaultType: XA
+   providerType: Atomikos
+   # When the provider type is Narayana, the following properties can be configured or not
+   props:
+     recoveryStoreUrl: jdbc:mysql://127.0.0.1:3306/jbossts
+     recoveryStoreDataSource: com.mysql.jdbc.jdbc2.optional.MysqlDataSource
+     recoveryStoreUser: root
+     recoveryStorePassword: 12345678
+ - !SQL_PARSER
+   sqlCommentParseEnabled: true
+   sqlStatementCache:
+     initialCapacity: 2000
+     maximumSize: 65535
+   parseTreeCache:
+     initialCapacity: 128
+     maximumSize: 1024
+
+props:
+  max-connections-size-per-query: 1
+  kernel-executor-size: 16  # Infinite by default.
+  proxy-frontend-flush-threshold: 128  # The default value is 128.
+  proxy-hint-enabled: true
+  sql-show: false
+  check-table-metadata-enabled: false
+  show-process-list-enabled: false
+    # Proxy backend query fetch size. A larger value may increase the memory usage of ShardingSphere Proxy.
+    # The default value is -1, which means set the minimum value for different JDBC drivers.
+  proxy-backend-query-fetch-size: -1
+  check-duplicate-table-enabled: false
+  proxy-frontend-executor-size: 0 # Proxy frontend executor size. The default value is 0, which means let Netty decide.
+    # Available options of proxy backend executor suitable: OLAP(default), OLTP. The OLTP option may reduce time cost of writing packets to client, but it may increase the latency of SQL execution
+    # and block other clients if client connections are more than `proxy-frontend-executor-size`, especially executing slow SQL.
+  proxy-backend-executor-suitable: OLAP
+  proxy-frontend-max-connections: 0 # Less than or equal to 0 means no limitation.
+  sql-federation-enabled: false
+    # Available proxy backend driver type: JDBC (default), ExperimentalVertx
+  proxy-backend-driver-type: JDBC
+  proxy-mysql-default-version: 5.7.22 # In the absence of schema name, the default version will be used.
+  proxy-default-port: 3307 # Proxy default port.
+  proxy-netty-backlog: 1024 # Proxy netty backlog.
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/docker-compose-mysql.yml b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/docker-compose-mysql.yml
new file mode 100644
index 000000000000..e3227a2523f4
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/docker-compose-mysql.yml
@@ -0,0 +1,60 @@
+# 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.
+
+version: '2.1'
+
+services:
+  sharding-proxy:
+    image: apache/shardingsphere-proxy:${SS_VERSION}
+    volumes:
+      - ./download-mysql.sh:/opt/shardingsphere-proxy/download-mysql.sh
+      - ./conf-mysql:/opt/shardingsphere-proxy/conf
+    networks:
+      - e2e
+    expose:
+      - 3307
+    entrypoint: ['sh', '-c', '/opt/shardingsphere-proxy/download-mysql.sh /opt/shardingsphere-proxy/ext-lib && /opt/shardingsphere-proxy/bin/start.sh 3307 && tail -f /opt/shardingsphere-proxy/logs/stdout.log']
+    healthcheck:
+      test: ["CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/3307"]
+      interval: 5s
+      timeout: 60s
+      retries: 120
+    depends_on:
+      data-source-0:
+        condition: service_healthy
+      data-source-1:
+        condition: service_healthy
+
+  data-source-0: &ds
+    image: mysql/mysql-server:8.0.13
+    networks:
+      - e2e
+    expose:
+      - 3306
+    environment:
+      MYSQL_ROOT_PASSWORD: "root@1234"
+      MYSQL_DATABASE: "swtest"
+      MYSQL_ROOT_HOST: "%"
+    healthcheck:
+      test: [ "CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/3306" ]
+      interval: 5s
+      timeout: 60s
+      retries: 120
+
+  data-source-1:
+    <<: *ds
+
+networks:
+  e2e:
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/download-mysql.sh b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/download-mysql.sh
new file mode 100755
index 000000000000..7ffd5e876089
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/download-mysql.sh
@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+
+# 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.
+
+set -ex
+
+LIB_HOME=$1
+MYSQL_URL="https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/8.0.13/mysql-connector-java-8.0.13.jar"
+MYSQL_DRIVER="mysql-connector-java-8.0.13.jar"
+
+# ensure the curl command been installed
+if ! command -v curl &> /dev/null; then
+    apt update -y && apt install -y curl
+fi
+
+if ! curl -Lo "${LIB_HOME}/${MYSQL_DRIVER}" ${MYSQL_URL}; then
+    echo "Fail to download ${MYSQL_DRIVER}."
+    exit 1
+fi
diff --git a/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/log4j2.xml b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/log4j2.xml
new file mode 100644
index 000000000000..cd672826bb3f
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-shardingsphere-plugin/src/test/resources/log4j2.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+  -->
+
+<Configuration status="INFO">
+    <Appenders>
+        <Console name="Console" target="SYSTEM_OUT">
+            <PatternLayout charset="UTF-8" pattern="%d - %c -%-4r [%t] %-5p %x - %m%n"/>
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="INFO">
+            <AppenderRef ref="Console"/>
+        </Root>
+    </Loggers>
+</Configuration>