1. Provide InsertRequest and UpdateRequest interface for prepare persistence. (#3131)

2. Implement the ids query for H2 metrics DAO.

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java

@@ -40,7 +40,7 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
 
     private final Model model;
     private final MergeDataCache<Metrics> mergeDataCache;
-    private final IMetricsDAO<?, ?> metricsDAO;
+    private final IMetricsDAO metricsDAO;
     private final AbstractWorker<Metrics> nextAlarmWorker;
     private final AbstractWorker<ExportEvent> nextExportWorker;
     private final DataCarrier<Metrics> dataCarrier;

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IMetricsDAO.java

@@ -22,15 +22,16 @@ import java.io.IOException;
 import java.util.Map;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.library.client.request.*;
 
 /**
  * @author peng-yongsheng
  */
-public interface IMetricsDAO<INSERT, UPDATE> extends DAO {
+public interface IMetricsDAO extends DAO {
 
     Map<String, Metrics> get(Model model, Metrics[] metrics) throws IOException;
 
-    INSERT prepareBatchInsert(Model model, Metrics metrics) throws IOException;
+    InsertRequest prepareBatchInsert(Model model, Metrics metrics) throws IOException;
 
-    UPDATE prepareBatchUpdate(Model model, Metrics metrics) throws IOException;
+    UpdateRequest prepareBatchUpdate(Model model, Metrics metrics) throws IOException;
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java

@@ -26,9 +26,9 @@ import org.apache.skywalking.oap.server.library.module.Service;
 /**
  * @author peng-yongsheng
  */
-public interface StorageDAO<INSERT, UPDATE> extends Service {
+public interface StorageDAO extends Service {
 
-    IMetricsDAO<INSERT, UPDATE> newMetricsDao(StorageBuilder<Metrics> storageBuilder);
+    IMetricsDAO newMetricsDao(StorageBuilder<Metrics> storageBuilder);
 
     IRegisterDAO newRegisterDao(StorageBuilder<RegisterSource> storageBuilder);
 

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java

@@ -254,14 +254,14 @@ public class ElasticSearchClient implements Client {
         client.update(request);
     }
 
-    public IndexRequest prepareInsert(String indexName, String id, XContentBuilder source) {
+    public ElasticSearchInsertRequest prepareInsert(String indexName, String id, XContentBuilder source) {
         indexName = formatIndexName(indexName);
-        return new IndexRequest(indexName, TYPE, id).source(source);
+        return new ElasticSearchInsertRequest(indexName, TYPE, id).source(source);
     }
 
-    public UpdateRequest prepareUpdate(String indexName, String id, XContentBuilder source) {
+    public ElasticSearchUpdateRequest prepareUpdate(String indexName, String id, XContentBuilder source) {
         indexName = formatIndexName(indexName);
-        return new UpdateRequest(indexName, TYPE, id).doc(source);
+        return new ElasticSearchUpdateRequest(indexName, TYPE, id).doc(source);
     }
 
     public int delete(String indexName, String timeBucketColumnName, long endTimeBucket) throws IOException {

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchInsertRequest.java

+/*
+ * 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.library.client.elasticsearch;
+
+import org.apache.skywalking.oap.server.library.client.request.InsertRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ElasticSearchInsertRequest extends IndexRequest implements InsertRequest {
+
+    public ElasticSearchInsertRequest(String index, String type, String id) {
+        super(index, type, id);
+    }
+
+    @Override public ElasticSearchInsertRequest source(XContentBuilder sourceBuilder) {
+        super.source(sourceBuilder);
+        return this;
+    }
+}

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchUpdateRequest.java

+/*
+ * 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.library.client.elasticsearch;
+
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ElasticSearchUpdateRequest extends UpdateRequest implements org.apache.skywalking.oap.server.library.client.request.UpdateRequest {
+
+    public ElasticSearchUpdateRequest(String index, String type, String id) {
+        super(index, type, id);
+    }
+
+    @Override public ElasticSearchUpdateRequest doc(XContentBuilder source) {
+        super.doc(source);
+        return this;
+    }
+}

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java

@@ -31,7 +31,7 @@ import org.slf4j.*;
  * @author wusheng
  */
 public class JDBCHikariCPClient implements Client {
-    private final Logger logger = LoggerFactory.getLogger(JDBCHikariCPClient.class);
+    private static final Logger logger = LoggerFactory.getLogger(JDBCHikariCPClient.class);
 
     private HikariDataSource dataSource;
     private HikariConfig hikariConfig;

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/request/InsertRequest.java

+/*
+ * 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.library.client.request;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface InsertRequest extends PrepareRequest {
+}

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/request/PrepareRequest.java

+/*
+ * 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.library.client.request;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface PrepareRequest {
+}

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/request/UpdateRequest.java

+/*
+ * 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.library.client.request;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface UpdateRequest extends PrepareRequest {
+}

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/MetricsEsDAO.java

@@ -23,16 +23,14 @@ import java.util.*;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.storage.*;
 import org.apache.skywalking.oap.server.core.storage.model.Model;
-import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
-import org.elasticsearch.action.index.IndexRequest;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.*;
 import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 
 /**
  * @author peng-yongsheng
  */
-public class MetricsEsDAO extends EsDAO implements IMetricsDAO<IndexRequest, UpdateRequest> {
+public class MetricsEsDAO extends EsDAO implements IMetricsDAO {
 
     private final StorageBuilder<Metrics> storageBuilder;
 
@@ -57,13 +55,13 @@ public class MetricsEsDAO extends EsDAO implements IMetricsDAO<IndexRequest, Upd
         return result;
     }
 
-    @Override public IndexRequest prepareBatchInsert(Model model, Metrics metrics) throws IOException {
+    @Override public ElasticSearchInsertRequest prepareBatchInsert(Model model, Metrics metrics) throws IOException {
         XContentBuilder builder = map2builder(storageBuilder.data2Map(metrics));
         String modelName = TimeSeriesUtils.timeSeries(model, metrics.getTimeBucket());
         return getClient().prepareInsert(modelName, metrics.id(), builder);
     }
 
-    @Override public UpdateRequest prepareBatchUpdate(Model model, Metrics metrics) throws IOException {
+    @Override public ElasticSearchUpdateRequest prepareBatchUpdate(Model model, Metrics metrics) throws IOException {
         XContentBuilder builder = map2builder(storageBuilder.data2Map(metrics));
         String modelName = TimeSeriesUtils.timeSeries(model, metrics.getTimeBucket());
         return getClient().prepareUpdate(modelName, metrics.id(), builder);

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java

@@ -23,19 +23,17 @@ import org.apache.skywalking.oap.server.core.analysis.record.Record;
 import org.apache.skywalking.oap.server.core.register.RegisterSource;
 import org.apache.skywalking.oap.server.core.storage.*;
 import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.update.UpdateRequest;
 
 /**
  * @author peng-yongsheng
  */
-public class StorageEsDAO extends EsDAO implements StorageDAO<IndexRequest, UpdateRequest> {
+public class StorageEsDAO extends EsDAO implements StorageDAO {
 
     public StorageEsDAO(ElasticSearchClient client) {
         super(client);
     }
 
-    @Override public IMetricsDAO<IndexRequest, UpdateRequest> newMetricsDao(StorageBuilder<Metrics> storageBuilder) {
+    @Override public IMetricsDAO newMetricsDao(StorageBuilder<Metrics> storageBuilder) {
         return new MetricsEsDAO(getClient(), storageBuilder);
     }
 

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/ArrayParamBuilder.java

+/*
+ * 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;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ArrayParamBuilder {
+
+    public static String build(String[] values) {
+        StringBuilder param = new StringBuilder();
+        for (int i = 0; i < values.length; i++) {
+            param.append("'").append(values[i]).append("'");
+            if (i < values.length - 1) {
+                param.append(",");
+            }
+        }
+        return param.toString();
+    }
+}

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/SQLExecutor.java

@@ -18,20 +18,19 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.jdbc;
 
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
+import java.sql.*;
 import java.util.List;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.skywalking.oap.server.library.client.request.*;
+import org.slf4j.*;
 
 /**
  * A SQL executor.
  *
  * @author wusheng
  */
-public class SQLExecutor {
-    private final Logger logger = LoggerFactory.getLogger(SQLExecutor.class);
+public class SQLExecutor implements InsertRequest, UpdateRequest {
+
+    private static final Logger logger = LoggerFactory.getLogger(SQLExecutor.class);
 
     private String sql;
     private List<Object> param;

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricsDAO.java

@@ -19,7 +19,7 @@
 package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao;
 
 import java.io.IOException;
-import java.util.Map;
+import java.util.*;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.storage.*;
 import org.apache.skywalking.oap.server.core.storage.model.Model;
@@ -29,7 +29,7 @@ import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLExecutor;
 /**
  * @author wusheng
  */
-public class H2MetricsDAO extends H2SQLExecutor implements IMetricsDAO<SQLExecutor, SQLExecutor> {
+public class H2MetricsDAO extends H2SQLExecutor implements IMetricsDAO {
 
     private JDBCHikariCPClient h2Client;
     private StorageBuilder<Metrics> storageBuilder;
@@ -40,8 +40,19 @@ public class H2MetricsDAO extends H2SQLExecutor implements IMetricsDAO<SQLExecut
     }
 
     @Override public Map<String, Metrics> get(Model model, Metrics[] metrics) throws IOException {
-        //        return (Metrics)getByID(h2Client, model.getName(), metrics.id(), storageBuilder);
-        return null;
+        Map<String, Metrics> result = new HashMap<>();
+
+        String[] ids = new String[metrics.length];
+        for (int i = 0; i < metrics.length; i++) {
+            ids[i] = metrics[i].id();
+        }
+
+        List<StorageData> storageDataList = getByIDs(h2Client, model.getName(), ids, storageBuilder);
+
+        for (StorageData storageData : storageDataList) {
+            result.put(storageData.id(), (Metrics)storageData);
+        }
+        return result;
     }
 
     @Override public SQLExecutor prepareBatchInsert(Model model, Metrics metrics) throws IOException {

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2SQLExecutor.java

@@ -32,11 +32,33 @@ import org.apache.skywalking.oap.server.storage.plugin.jdbc.*;
 import org.slf4j.*;
 
 /**
- * @author wusheng
+ * @author wusheng, peng-yongsheng
  */
 public class H2SQLExecutor {
     private static final Logger logger = LoggerFactory.getLogger(H2SQLExecutor.class);
 
+    protected List<StorageData> getByIDs(JDBCHikariCPClient h2Client, String modelName, String[] ids,
+        StorageBuilder storageBuilder) throws IOException {
+
+        try (Connection connection = h2Client.getConnection()) {
+            /*
+             * Although H2 database or other database support createArrayOf and setArray operate.
+             * But Mysql 5.1.44 driver doesn't.
+             */
+            String param = ArrayParamBuilder.build(ids);
+
+            try (ResultSet rs = h2Client.executeQuery(connection, "SELECT * FROM " + modelName + " WHERE id in (" + param + ")")) {
+                List<StorageData> storageDataList = new ArrayList<>();
+                while (rs.next()) {
+                    storageDataList.add(toStorageData(rs, modelName, storageBuilder));
+                }
+                return storageDataList;
+            }
+        } catch (SQLException | JDBCClientException e) {
+            throw new IOException(e.getMessage(), e);
+        }
+    }
+
     protected StorageData getByID(JDBCHikariCPClient h2Client, String modelName, String id,
         StorageBuilder storageBuilder) throws IOException {
         try (Connection connection = h2Client.getConnection()) {
@@ -59,8 +81,7 @@ public class H2SQLExecutor {
         }
     }
 
-    protected StorageData toStorageData(ResultSet rs, String modelName,
-        StorageBuilder storageBuilder) throws SQLException {
+    protected StorageData toStorageData(ResultSet rs, String modelName, StorageBuilder storageBuilder) throws SQLException {
         if (rs.next()) {
             Map data = new HashMap();
             List<ModelColumn> columns = TableMetaInfo.get(modelName).getColumns();
@@ -85,8 +106,7 @@ public class H2SQLExecutor {
         return Const.NONE;
     }
 
-    protected SQLExecutor getInsertExecutor(String modelName, StorageData metrics,
-        StorageBuilder storageBuilder) throws IOException {
+    protected SQLExecutor getInsertExecutor(String modelName, StorageData metrics, StorageBuilder storageBuilder) throws IOException {
         Map<String, Object> objectMap = storageBuilder.data2Map(metrics);
 
         SQLBuilder sqlBuilder = new SQLBuilder("INSERT INTO " + modelName + " VALUES");
@@ -113,8 +133,7 @@ public class H2SQLExecutor {
         return new SQLExecutor(sqlBuilder.toString(), param);
     }
 
-    protected SQLExecutor getUpdateExecutor(String modelName, StorageData metrics,
-        StorageBuilder storageBuilder) throws IOException {
+    protected SQLExecutor getUpdateExecutor(String modelName, StorageData metrics, StorageBuilder storageBuilder) throws IOException {
         Map<String, Object> objectMap = storageBuilder.data2Map(metrics);
 
         SQLBuilder sqlBuilder = new SQLBuilder("UPDATE " + modelName + " SET ");

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2StorageDAO.java

@@ -23,12 +23,11 @@ import org.apache.skywalking.oap.server.core.analysis.record.Record;
 import org.apache.skywalking.oap.server.core.register.RegisterSource;
 import org.apache.skywalking.oap.server.core.storage.*;
 import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient;
-import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLExecutor;
 
 /**
- * @author wusheng
+ * @author wusheng, peng-yongsheng
  */
-public class H2StorageDAO implements StorageDAO<SQLExecutor, SQLExecutor> {
+public class H2StorageDAO implements StorageDAO {
 
     private JDBCHikariCPClient h2Client;
 
@@ -36,7 +35,7 @@ public class H2StorageDAO implements StorageDAO<SQLExecutor, SQLExecutor> {
         this.h2Client = h2Client;
     }
 
-    @Override public IMetricsDAO<SQLExecutor, SQLExecutor> newMetricsDao(StorageBuilder<Metrics> storageBuilder) {
+    @Override public IMetricsDAO newMetricsDao(StorageBuilder<Metrics> storageBuilder) {
         return new H2MetricsDAO(h2Client, storageBuilder);
     }
 

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/ArrayParamBuilderTestCase.java

+/*
+ * 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;
+
+import org.junit.*;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ArrayParamBuilderTestCase {
+
+    @Test
+    public void testBuild() {
+        String param = ArrayParamBuilder.build(new String[] {"1"});
+        Assert.assertEquals("'1'", param);
+
+        param = ArrayParamBuilder.build(new String[] {"1", "2"});
+        Assert.assertEquals("'1','2'", param);
+
+        param = ArrayParamBuilder.build(new String[] {"1", "2", "3"});
+        Assert.assertEquals("'1','2','3'", param);
+    }
+}