This is an automated email from the ASF dual-hosted git repository. lta pushed a commit to branch reimpl_sync in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 0d96f69b3fc19ffc96a696017b0b80c4867c883f Author: lta <[email protected]> AuthorDate: Sun Aug 25 19:35:46 2019 +0800 update --- .../db/sync/receiver/transfer/SyncServiceImpl.java | 31 ++- .../db/sync/sender/MultipleClientSyncTest.java | 226 ----------------- .../org/apache/iotdb/db/sync/test/RandomNum.java | 70 ----- .../apache/iotdb/db/sync/test/SyncTestClient1.java | 258 ------------------- .../apache/iotdb/db/sync/test/SyncTestClient2.java | 262 ------------------- .../apache/iotdb/db/sync/test/SyncTestClient3.java | 282 --------------------- .../java/org/apache/iotdb/db/sync/test/Utils.java | 44 ---- 7 files changed, 21 insertions(+), 1152 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java index e2149b6..7156ef7 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java @@ -67,8 +67,6 @@ public class SyncServiceImpl implements SyncService.Iface { private ThreadLocal<String> syncFolderPath = new ThreadLocal<>(); - private ThreadLocal<String> syncDataPath = new ThreadLocal<>(); - private ThreadLocal<String> currentSG = new ThreadLocal<>(); private ThreadLocal<SyncReceiverLogger> syncLog = new ThreadLocal<>(); @@ -101,7 +99,15 @@ public class SyncServiceImpl implements SyncService.Iface { } private boolean checkRecovery() { - return true; + try { + if (currentFileWriter.get() != null && currentFileWriter.get().isOpen()) { + currentFileWriter.get().close(); + } + return true; + } catch (IOException e) { + e.printStackTrace(); + return false; + } } @Override @@ -109,7 +115,7 @@ public class SyncServiceImpl implements SyncService.Iface { try { initPath(); currentSG.remove(); - syncLog.set(new SyncReceiverLogger(new File(syncDataPath.get(), Constans.SYNC_LOG_NAME))); + syncLog.set(new SyncReceiverLogger(new File(getSyncDataPath(), Constans.SYNC_LOG_NAME))); return getSuccessResult(); } catch (DiskSpaceInsufficientException | IOException e) { logger.error("Can not receiver data from sender", e); @@ -125,8 +131,6 @@ public class SyncServiceImpl implements SyncService.Iface { syncFolderPath .set(FilePathUtils.regularizePath(dataDir) + Constans.SYNC_RECEIVER + File.separatorChar + senderIp.get()); - syncDataPath - .set(syncFolderPath.get() + File.separatorChar + Constans.RECEIVER_DATA_FOLDER_NAME); } /** @@ -163,14 +167,17 @@ public class SyncServiceImpl implements SyncService.Iface { try { File file; if (currentSG.get() == null) { - file = new File(syncDataPath.get(), filename); + file = new File(getSyncDataPath(), filename); } else { - file = new File(syncDataPath.get(), currentSG.get() + File.separatorChar + filename); + file = new File(getSyncDataPath(), currentSG.get() + File.separatorChar + filename); } currentFile.set(file); if (!file.getParentFile().exists()) { file.getParentFile().mkdirs(); } + if (currentFileWriter.get() != null && currentFileWriter.get().isOpen()) { + currentFileWriter.get().close(); + } currentFileWriter.set(new FileOutputStream(file).getChannel()); syncLog.get().startSyncTsFiles(); } catch (IOException e) { @@ -294,14 +301,18 @@ public class SyncServiceImpl implements SyncService.Iface { public ResultStatus endSync() throws TException { try { syncLog.get().close(); - new File(syncDataPath.get(), Constans.SYNC_END).createNewFile(); + new File(getSyncDataPath(), Constans.SYNC_END).createNewFile(); } catch (IOException e) { logger.error("Can not end sync", e); - return getErrorResult(String.format("Can not end sync because {}", e.getMessage())); + return getErrorResult(String.format("Can not end sync because %s", e.getMessage())); } return getSuccessResult(); } + private String getSyncDataPath() { + return syncFolderPath.get() + File.separatorChar + Constans.RECEIVER_DATA_FOLDER_NAME; + } + private ResultStatus getSuccessResult() { return new ResultStatus(true, null, null); } diff --git a/server/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientSyncTest.java b/server/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientSyncTest.java deleted file mode 100644 index 52a7138..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientSyncTest.java +++ /dev/null @@ -1,226 +0,0 @@ -/** - * 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.iotdb.db.sync.sender; - -import static org.junit.Assert.fail; - -import java.io.BufferedReader; -import java.io.File; -import java.io.FileReader; -import java.io.IOException; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import org.apache.iotdb.jdbc.Config; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class MultipleClientSyncTest { - - Map<String, ArrayList<String>> timeseriesList = new HashMap(); - Map<String, ArrayList<String>> timeseriesList1 = new HashMap(); - private static final Logger logger = LoggerFactory.getLogger(MultipleClientSyncTest.class); - private Set<String> dataSender = new HashSet<>(); - private Set<String> dataReceiver = new HashSet<>(); - - public static void main(String[] args) throws IOException { - MultipleClientSyncTest multipleClientSyncTest = new MultipleClientSyncTest(); - multipleClientSyncTest.testPostback(); - } - - public void testPostback() throws IOException { - - timeseriesList1.put("root.vehicle_history1", new ArrayList<>()); - timeseriesList1.put("root.vehicle_alarm1", new ArrayList<>()); - timeseriesList1.put("root.vehicle_temp1", new ArrayList<>()); - timeseriesList1.put("root.range_event1", new ArrayList<>()); - timeseriesList.put("root.vehicle_history", new ArrayList<>()); - timeseriesList.put("root.vehicle_alarm", new ArrayList<>()); - timeseriesList.put("root.vehicle_temp", new ArrayList<>()); - timeseriesList.put("root.range_event", new ArrayList<>()); - - File file = new File("CreateTimeseries1.txt"); - BufferedReader reader = new BufferedReader(new FileReader(file)); - String line; - while ((line = reader.readLine()) != null) { - String timeseries = line.split(" ")[2]; - for (String storageGroup : timeseriesList.keySet()) { - if (timeseries.startsWith(storageGroup + ".")) { - String timesery = timeseries.substring((storageGroup + ".").length()); - timeseriesList.get(storageGroup).add(timesery); - break; - } - } - } - - file = new File("CreateTimeseries2.txt"); - reader = new BufferedReader(new FileReader(file)); - while ((line = reader.readLine()) != null) { - String timeseries = line.split(" ")[2]; - for (String storageGroup : timeseriesList1.keySet()) { - if (timeseries.startsWith(storageGroup + ".")) { - String timesery = timeseries.substring((storageGroup + ".").length()); - timeseriesList1.get(storageGroup).add(timesery); - break; - } - } - } - - for (String storageGroup : timeseriesList.keySet()) { - String sqlFormat = "select %s from %s"; - logger.debug(String.format("%s:", storageGroup)); - int count = 0; - int count1 = 0; - int count2 = 0; - for (String timesery : timeseriesList.get(storageGroup)) { - count++; - count1 = 0; - count2 = 0; - dataSender.clear(); - dataReceiver.clear(); - try { - Class.forName(Config.JDBC_DRIVER_NAME); - Connection connection = null; - Connection connection1 = null; - try { - connection = DriverManager - .getConnection("jdbc:iotdb://192.168.130.14:6667/", "root", "root"); - connection1 = DriverManager - .getConnection("jdbc:iotdb://192.168.130.16:6667/", "root", "root"); - Statement statement = connection.createStatement(); - Statement statement1 = connection1.createStatement(); - String sql = String.format(sqlFormat, timesery, storageGroup); - boolean hasResultSet = statement.execute(sql); - boolean hasResultSet1 = statement1.execute(sql); - if (hasResultSet) { - ResultSet res = statement.getResultSet(); - while (res.next()) { - count1++; - dataSender - .add(res.getString("Time") + res.getString(storageGroup + "." + timesery)); - } - } - if (hasResultSet1) { - ResultSet res = statement1.getResultSet(); - while (res.next()) { - count2++; - dataReceiver - .add(res.getString("Time") + res.getString(storageGroup + "." + timesery)); - } - } - assert ((dataSender.size() == dataReceiver.size()) && dataSender - .containsAll(dataReceiver)); - statement.close(); - statement1.close(); - } catch (Exception e) { - logger.error("", e); - } finally { - if (connection != null) { - connection.close(); - } - if (connection1 != null) { - connection1.close(); - } - } - } catch (ClassNotFoundException | SQLException e) { - fail(e.getMessage()); - } - if (count > 20) { - break; - } - logger.debug(String.valueOf(count1)); - logger.debug(String.valueOf(count2)); - } - } - - for (String storageGroup : timeseriesList1.keySet()) { - String sqlFormat = "select %s from %s"; - logger.debug(String.format("%s:", storageGroup)); - int count = 0; - int count1; - int count2; - for (String timesery : timeseriesList1.get(storageGroup)) { - count++; - count1 = 0; - count2 = 0; - dataSender.clear(); - dataReceiver.clear(); - try { - Class.forName(Config.JDBC_DRIVER_NAME); - Connection connection = null; - Connection connection1 = null; - try { - connection = DriverManager - .getConnection("jdbc:iotdb://192.168.130.15:6667/", "root", "root"); - connection1 = DriverManager - .getConnection("jdbc:iotdb://192.168.130.16:6667/", "root", "root"); - Statement statement = connection.createStatement(); - Statement statement1 = connection1.createStatement(); - String sql = String.format(sqlFormat, timesery, storageGroup); - boolean hasResultSet = statement.execute(sql); - boolean hasResultSet1 = statement1.execute(sql); - if (hasResultSet) { - ResultSet res = statement.getResultSet(); - while (res.next()) { - count1++; - dataSender - .add(res.getString("Time") + res.getString(storageGroup + "." + timesery)); - } - } - if (hasResultSet1) { - ResultSet res = statement1.getResultSet(); - while (res.next()) { - count2++; - dataReceiver - .add(res.getString("Time") + res.getString(storageGroup + "." + timesery)); - } - } - assert ((dataSender.size() == dataReceiver.size()) && dataSender - .containsAll(dataReceiver)); - statement.close(); - statement1.close(); - } catch (Exception e) { - logger.error("", e); - } finally { - if (connection != null) { - connection.close(); - } - if (connection1 != null) { - connection1.close(); - } - } - } catch (ClassNotFoundException | SQLException e) { - fail(e.getMessage()); - } - if (count > 20) { - break; - } - logger.debug(String.valueOf(count1)); - logger.debug(String.valueOf(count2)); - } - } - } -} diff --git a/server/src/test/java/org/apache/iotdb/db/sync/test/RandomNum.java b/server/src/test/java/org/apache/iotdb/db/sync/test/RandomNum.java deleted file mode 100644 index 125e9d3..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/test/RandomNum.java +++ /dev/null @@ -1,70 +0,0 @@ -/** - * 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.iotdb.db.sync.test; - -import java.util.Random; - -public class RandomNum { - - private static Random random = new Random(); - - private RandomNum() { - throw new IllegalStateException("Utility class"); - } - - public static long getRandomLong(long min, long max) { - return random.nextLong() % (max - min + 1) + min; - } - - public static int getRandomInt(int min, int max) { - return (random.nextInt(10000) % (max - min) + min); - } - - /** - * get random float between min and max. - */ - public static float getRandomFloat(float min, float max) { - - return (random.nextFloat() * (max - min) + min); - } - - /** - * get random int between 0 and frequency. - */ - public static int getAbnormalData(int frequency) { - return random.nextInt() % frequency; - } - - /** - * get random text consisting of lowercase letters and numbers. - * - * @param length -the size of random text - */ - public static String getRandomText(int length) { - - String base = "abcdefghijklmnopqrstuvwxyz0123456789"; - StringBuilder st = new StringBuilder(); - for (int i = 0; i < length; i++) { - int number = random.nextInt(base.length()); - st = st.append(base.charAt(number)); - } - return st.toString(); - - } -} \ No newline at end of file diff --git a/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient1.java b/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient1.java deleted file mode 100644 index e7e7b9e..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient1.java +++ /dev/null @@ -1,258 +0,0 @@ -/** - * 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.iotdb.db.sync.test; - -import static org.apache.iotdb.db.sync.test.RandomNum.getRandomInt; - -import java.io.BufferedReader; -import java.io.File; -import java.io.FileReader; -import java.io.IOException; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import org.apache.iotdb.db.conf.IoTDBConstant; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * SyncTestClient1 is used to generate data of whole timeseries (simulating jilian scene) to test stability of - * sync function. - */ -public class SyncTestClient1 { - - private static final int TIME_INTERVAL = 0; - private static final int TOTAL_DATA = 2000000; - private static final int ABNORMAL_MAX_INT = 0; - private static final int ABNORMAL_MIN_INT = -10; - private static final int ABNORMAL_MAX_FLOAT = 0; - private static final int ABNORMAL_MIN_FLOAT = -10; - private static final int ABNORMAL_FREQUENCY = Integer.MAX_VALUE; - private static final int ABNORMAL_LENGTH = 0; - private static final int MIN_INT = 0; - private static final int MAX_INT = 14; - private static final int MIN_FLOAT = 20; - private static final int MAX_FLOAT = 30; - private static final int STRING_LENGTH = 5; - private static final int BATCH_SQL = 10000; - private static final Logger logger = LoggerFactory.getLogger(SyncTestClient1.class); - - /** - * generate time series map from file. - * - * @param inputFilePath input file path - * @return Map - * @throws Exception Exception - */ - public static Map<String, String> generateTimeseriesMapFromFile(String inputFilePath) - throws IOException { - - Map<String, String> timeseriesMap = new HashMap<>(); - - File file = new File(inputFilePath); - try (BufferedReader reader = new BufferedReader(new FileReader(file))) { - String line; - while ((line = reader.readLine()) != null) { - - String timeseries = line.split(" ")[2]; - String dataType = line.split("DATATYPE = ")[1].split(",")[0].trim(); - String encodingType = line.split("ENCODING = ")[1].split(";")[0].trim(); - timeseriesMap.put(timeseries, dataType + "," + encodingType); - } - } - return timeseriesMap; - - } - - /** - * create time series. - * - * @param statement statement - * @param timeseriesMap time series map - */ - public static void createTimeseries(Statement statement, Map<String, String> timeseriesMap) { - - try { - String createTimeseriesSql = "CREATE TIMESERIES <timeseries> WITH DATATYPE=<datatype>, " - + "ENCODING=<encode>"; - - int sqlCount = 0; - for (Map.Entry<String, String> entry : timeseriesMap.entrySet()) { - String key = entry.getKey(); - String properties = entry.getValue(); - String sql = createTimeseriesSql.replace("<timeseries>", key) - .replace("<datatype>", Utils.getType(properties)) - .replace("<encode>", Utils.getEncode(properties)); - statement.addBatch(sql); - sqlCount++; - if (sqlCount >= BATCH_SQL) { - statement.executeBatch(); - statement.clearBatch(); - sqlCount = 0; - } - } - statement.executeBatch(); - statement.clearBatch(); - } catch (Exception e) { - logger.error("", e); - } - } - - /** - * set storage group. - * - * @param statement statement - * @param storageGroupList storage group list - * @throws SQLException SQLException - */ - public static void setStorageGroup(Statement statement, List<String> storageGroupList) - throws SQLException { - try { - String setStorageGroupSql = "SET STORAGE GROUP TO <prefixpath>"; - for (String str : storageGroupList) { - String sql = setStorageGroupSql.replace("<prefixpath>", str); - statement.execute(sql); - } - } catch (Exception e) { - logger.error("", e); - } - } - - /** - * random insert data. - * - * @param statement statement - * @param timeseriesMap time series map - * @throws Exception Exception - */ - public static void randomInsertData(Statement statement, Map<String, String> timeseriesMap) - throws SQLException, InterruptedException { - String insertDataSql = "INSERT INTO %s (timestamp, %s) VALUES (%s, %s)"; - int abnormalCount = 0; - int abnormalFlag = 1; - int sqlCount = 0; - - for (int i = 0; i < TOTAL_DATA; i++) { - - long time = System.currentTimeMillis(); - - if (i % ABNORMAL_FREQUENCY == 250) { - abnormalFlag = 0; - } - - for (Map.Entry<String, String> entry : timeseriesMap.entrySet()) { - String key = entry.getKey(); - String type = Utils.getType(entry.getValue()); - String path = Utils.getPath(key); - String sensor = Utils.getSensor(key); - String sql = ""; - - if (type.equals("INT32")) { - int value; - if (abnormalFlag == 0) { - value = getRandomInt(ABNORMAL_MIN_INT, ABNORMAL_MAX_INT); - } else { - value = getRandomInt(MIN_INT, MAX_INT); - } - sql = String.format(insertDataSql, path, sensor, time, value); - } else if (type.equals("FLOAT")) { - float value; - if (abnormalFlag == 0) { - value = RandomNum.getRandomFloat(ABNORMAL_MIN_FLOAT, ABNORMAL_MAX_FLOAT); - } else { - value = RandomNum.getRandomFloat(MIN_FLOAT, MAX_FLOAT); - } - sql = String.format(insertDataSql, path, sensor, time, value); - } else if (type.equals("TEXT")) { - String value; - value = RandomNum.getRandomText(STRING_LENGTH); - sql = String.format(insertDataSql, path, sensor, time, "\"" + value + "\""); - } - - statement.addBatch(sql); - sqlCount++; - if (sqlCount >= BATCH_SQL) { - statement.executeBatch(); - statement.clearBatch(); - sqlCount = 0; - } - } - - if (abnormalFlag == 0) { - abnormalCount += 1; - } - if (abnormalCount >= ABNORMAL_LENGTH) { - abnormalCount = 0; - abnormalFlag = 1; - } - } - statement.executeBatch(); - statement.clearBatch(); - } - - /** - * main function. - * - * @param args arguments - * @throws Exception Exception - */ - public static void main(String[] args) throws Exception { - - Statement statement = null; - - String path = - new File(System.getProperty(IoTDBConstant.IOTDB_HOME, null)).getParent() + File.separator - + "src" - + File.separator + "test" + File.separator + "resources" + File.separator - + "CreateTimeseries1.txt"; - Map<String, String> timeseriesMap = generateTimeseriesMapFromFile(path); - - List<String> storageGroupList = new ArrayList<>(); - storageGroupList.add("root.vehicle_history"); - storageGroupList.add("root.vehicle_alarm"); - storageGroupList.add("root.vehicle_temp"); - storageGroupList.add("root.range_event"); - - try (Connection connection = DriverManager - .getConnection("jdbc:iotdb://localhost:6667/", "root", "root")) { - Class.forName("org.apache.iotdb.jdbc.IoTDBDriver"); - statement = connection.createStatement(); - - setStorageGroup(statement, storageGroupList); - logger.debug("Finish set storage group."); - createTimeseries(statement, timeseriesMap); - logger.debug("Finish create timeseries."); - while (true) { - randomInsertData(statement, timeseriesMap); - } - - } catch (Exception e) { - logger.error("", e); - } finally { - if (statement != null) { - statement.close(); - } - } - } -} \ No newline at end of file diff --git a/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient2.java b/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient2.java deleted file mode 100644 index 8f4de9f..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient2.java +++ /dev/null @@ -1,262 +0,0 @@ -/** - * 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.iotdb.db.sync.test; - -import java.io.BufferedReader; -import java.io.File; -import java.io.FileReader; -import java.io.IOException; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import org.apache.iotdb.db.conf.IoTDBConstant; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * SyncTestClient2 is used to generate data of half timeseries (simulating jilian scene) to test stability of - * sync function. - */ -public class SyncTestClient2 { - - private static final int TIME_INTERVAL = 0; - private static final int TOTAL_DATA = 2000000; - private static final int ABNORMAL_MAX_INT = 0; - private static final int ABNORMAL_MIN_INT = -10; - private static final int ABNORMAL_MAX_FLOAT = 0; - private static final int ABNORMAL_MIN_FLOAT = -10; - private static final int ABNORMAL_FREQUENCY = Integer.MAX_VALUE; - private static final int ABNORMAL_LENGTH = 0; - private static final int MIN_INT = 0; - private static final int MAX_INT = 14; - private static final int MIN_FLOAT = 20; - private static final int MAX_FLOAT = 30; - private static final int STRING_LENGTH = 5; - private static final int BATCH_SQL = 10000; - private static final Logger logger = LoggerFactory.getLogger(SyncTestClient2.class); - - /** - * generate time series map from file. - * - * @param inputFilePath input file path - * @return map - * @throws Exception Exception - */ - public static Map<String, String> generateTimeseriesMapFromFile(String inputFilePath) - throws IOException { - - Map<String, String> timeseriesMap = new HashMap<>(); - - File file = new File(inputFilePath); - try (BufferedReader reader = new BufferedReader(new FileReader(file))) { - String line; - while ((line = reader.readLine()) != null) { - - String timeseries = line.split(" ")[2]; - String dataType = line.split("DATATYPE = ")[1].split(",")[0].trim(); - String encodingType = line.split("ENCODING = ")[1].split(";")[0].trim(); - timeseriesMap.put(timeseries, dataType + "," + encodingType); - } - } - - return timeseriesMap; - - } - - /** - * create time series. - * - * @param statement statement - * @param timeseriesMap time series map - * @throws SQLException SQLException - */ - public static void createTimeseries(Statement statement, Map<String, String> timeseriesMap) - throws SQLException { - - try { - String createTimeseriesSql = "CREATE TIMESERIES <timeseries> WITH DATATYPE=<datatype>, " - + "ENCODING=<encode>"; - - int sqlCount = 0; - - for (Map.Entry<String, String> entry : timeseriesMap.entrySet()) { - String key = entry.getKey(); - String properties = entry.getValue(); - String sql = createTimeseriesSql.replace("<timeseries>", key) - .replace("<datatype>", Utils.getType(properties)) - .replace("<encode>", Utils.getEncode(properties)); - - statement.addBatch(sql); - sqlCount++; - if (sqlCount >= BATCH_SQL) { - statement.executeBatch(); - statement.clearBatch(); - sqlCount = 0; - } - } - statement.executeBatch(); - statement.clearBatch(); - } catch (Exception e) { - logger.error("", e); - } - } - - /** - * set storage group. - * - * @param statement statement - * @param storageGroupList storage group list - * @throws SQLException SQLException - */ - public static void setStorageGroup(Statement statement, List<String> storageGroupList) - throws SQLException { - - try { - String setStorageGroupSql = "SET STORAGE GROUP TO <prefixpath>"; - for (String str : storageGroupList) { - String sql = setStorageGroupSql.replace("<prefixpath>", str); - statement.execute(sql); - } - } catch (Exception e) { - logger.error("", e); - } - } - - /** - * randomly insert data. - * - * @param statement statement - * @param timeseriesMap time series map - */ - public static void randomInsertData(Statement statement, Map<String, String> timeseriesMap) - throws SQLException, InterruptedException { - String insertDataSql = "INSERT INTO %s (timestamp, %s) VALUES (%s, %s)"; - int abnormalCount = 0; - int abnormalFlag = 1; - - int sqlCount = 0; - - for (int i = 0; i < TOTAL_DATA; i++) { - - long time = System.currentTimeMillis(); - - if (i % ABNORMAL_FREQUENCY == 250) { - abnormalFlag = 0; - } - - for (Map.Entry<String, String> entry : timeseriesMap.entrySet()) { - String key = entry.getKey(); - String type = Utils.getType(entry.getValue()); - String path = Utils.getPath(key); - String sensor = Utils.getSensor(key); - String sql = ""; - - if (type.equals("INT32")) { - int value; - if (abnormalFlag == 0) { - value = RandomNum.getRandomInt(ABNORMAL_MIN_INT, ABNORMAL_MAX_INT); - } else { - value = RandomNum.getRandomInt(MIN_INT, MAX_INT); - } - sql = String.format(insertDataSql, path, sensor, time, value); - } else if (type.equals("FLOAT")) { - float value; - if (abnormalFlag == 0) { - value = RandomNum.getRandomFloat(ABNORMAL_MIN_FLOAT, ABNORMAL_MAX_FLOAT); - } else { - value = RandomNum.getRandomFloat(MIN_FLOAT, MAX_FLOAT); - } - sql = String.format(insertDataSql, path, sensor, time,value); - } else if (type.equals("TEXT")) { - String value; - value = RandomNum.getRandomText(STRING_LENGTH); - sql = String.format(insertDataSql, path, sensor, time, "\"" + value + "\""); - } - - statement.addBatch(sql); - sqlCount++; - if (sqlCount >= BATCH_SQL) { - statement.executeBatch(); - statement.clearBatch(); - sqlCount = 0; - } - } - - if (abnormalFlag == 0) { - abnormalCount += 1; - } - if (abnormalCount >= ABNORMAL_LENGTH) { - abnormalCount = 0; - abnormalFlag = 1; - } - } - statement.executeBatch(); - statement.clearBatch(); - } - - /** - * main function. - * - * @param args arguments - * @throws Exception Exception - */ - public static void main(String[] args) throws Exception { - - Statement statement = null; - - String path = - new File(System.getProperty(IoTDBConstant.IOTDB_HOME, null)).getParent() + File.separator - + "src" - + File.separator + "test" + File.separator + "resources" + File.separator - + "CreateTimeseries2.txt"; - Map<String, String> timeseriesMap = generateTimeseriesMapFromFile(path); - - List<String> storageGroupList = new ArrayList<>(); - storageGroupList.add("root.vehicle_history1"); - storageGroupList.add("root.vehicle_alarm1"); - storageGroupList.add("root.vehicle_temp1"); - storageGroupList.add("root.range_event1"); - - try (Connection connection = DriverManager - .getConnection("jdbc:iotdb://localhost:6667/", "root", "root")) { - Class.forName("org.apache.iotdb.jdbc.IoTDBDriver"); - statement = connection.createStatement(); - - setStorageGroup(statement, storageGroupList); - logger.debug("Finish set storage group."); - createTimeseries(statement, timeseriesMap); - logger.debug("Finish create timeseries."); - while (true) { - randomInsertData(statement, timeseriesMap); - } - - } catch (Exception e) { - logger.error("", e); - } finally { - if (statement != null) { - statement.close(); - } - } - } -} \ No newline at end of file diff --git a/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient3.java b/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient3.java deleted file mode 100644 index 5cfd935..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient3.java +++ /dev/null @@ -1,282 +0,0 @@ -/** - * 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.iotdb.db.sync.test; - -import java.io.BufferedReader; -import java.io.File; -import java.io.FileReader; -import java.io.IOException; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import org.apache.iotdb.db.conf.IoTDBConstant; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * SyncTestClient3 is used to generate data of another half timeseries (simulating jilian scene) which is - * different to those in SyncTestClient2 to test stability of sync function. - */ -public class SyncTestClient3 { - - private static final int TIME_INTERVAL = 0; - private static final int TOTAL_DATA = 2000000; - private static final int ABNORMAL_MAX_INT = 0; - private static final int ABNORMAL_MIN_INT = -10; - private static final int ABNORMAL_MAX_FLOAT = 0; - private static final int ABNORMAL_MIN_FLOAT = -10; - private static final int ABNORMAL_FREQUENCY = Integer.MAX_VALUE; - private static final int ABNORMAL_LENGTH = 0; - private static final int MIN_INT = 0; - private static final int MAX_INT = 14; - private static final int MIN_FLOAT = 20; - private static final int MAX_FLOAT = 30; - private static final int STRING_LENGTH = 5; - private static final int BATCH_SQL = 10000; - private static final Logger logger = LoggerFactory.getLogger(SyncTestClient3.class); - - /** - * generate time series map from file. - * - * @param inputFilePath input file path - * @return map - * @throws Exception Exception - */ - public static Map<String, String> generateTimeseriesMapFromFile(String inputFilePath) - throws IOException { - - Map<String, String> timeseriesMap = new HashMap<>(); - - File file = new File(inputFilePath); - try (BufferedReader reader = new BufferedReader(new FileReader(file))) { - String line; - while ((line = reader.readLine()) != null) { - String timeseries = line.split(" ")[2]; - String dataType = line.split("DATATYPE = ")[1].split(",")[0].trim(); - String encodingType = line.split("ENCODING = ")[1].split(";")[0].trim(); - timeseriesMap.put(timeseries, dataType + "," + encodingType); - } - } - - return timeseriesMap; - - } - - /** - * create time series. - * - * @param statement statement - * @param timeseriesMap time series map - * @throws SQLException SQLException - */ - public static void createTimeseries(Statement statement, Statement statement1, - Map<String, String> timeseriesMap) - throws SQLException { - - try { - String createTimeseriesSql = "CREATE TIMESERIES <timeseries> WITH DATATYPE=<datatype>, " - + "ENCODING=<encode>"; - - int sqlCount = 0; - - for (Map.Entry<String, String> entry : timeseriesMap.entrySet()) { - String key = entry.getKey(); - String properties = entry.getValue(); - String sql = createTimeseriesSql.replace("<timeseries>", key) - .replace("<datatype>", Utils.getType(properties)) - .replace("<encode>", Utils.getEncode(properties)); - - statement.addBatch(sql); - statement1.addBatch(sql); - sqlCount++; - if (sqlCount >= BATCH_SQL) { - statement.executeBatch(); - statement.clearBatch(); - statement1.executeBatch(); - statement1.clearBatch(); - sqlCount = 0; - } - } - statement.executeBatch(); - statement.clearBatch(); - statement1.executeBatch(); - statement1.clearBatch(); - } catch (Exception e) { - logger.error("", e); - } - } - - /** - * set storage group. - * - * @param statement statement - * @param storageGroupList storage group list - * @throws SQLException SQLException - */ - public static void setStorageGroup(Statement statement, Statement statement1, - List<String> storageGroupList) - throws SQLException { - try { - String setStorageGroupSql = "SET STORAGE GROUP TO <prefixpath>"; - for (String str : storageGroupList) { - String sql = setStorageGroupSql.replace("<prefixpath>", str); - statement.execute(sql); - statement1.execute(sql); - } - } catch (Exception e) { - logger.error("", e); - } - } - - /** - * randomly insert data. - * - * @param statement statement - * @param timeseriesMap time series map - * @throws Exception Exception - */ - public static void randomInsertData(Statement statement, Statement statement1, - Map<String, String> timeseriesMap) throws InterruptedException, SQLException { - String insertDataSql = "INSERT INTO %s (timestamp, %s) VALUES (%s, %s)"; - int abnormalCount = 0; - int abnormalFlag = 1; - - int sqlCount = 0; - - for (int i = 0; i < TOTAL_DATA; i++) { - - long time = System.currentTimeMillis(); - - if (i % ABNORMAL_FREQUENCY == 250) { - abnormalFlag = 0; - } - - for (Map.Entry<String, String> entry : timeseriesMap.entrySet()) { - String key = entry.getKey(); - String type = Utils.getType(entry.getValue()); - String path = Utils.getPath(key); - String sensor = Utils.getSensor(key); - String sql = ""; - - if (type.equals("INT32")) { - int value; - if (abnormalFlag == 0) { - value = RandomNum.getRandomInt(ABNORMAL_MIN_INT, ABNORMAL_MAX_INT); - } else { - value = RandomNum.getRandomInt(MIN_INT, MAX_INT); - } - sql = String.format(insertDataSql, path, sensor, time, value); - } else if (type.equals("FLOAT")) { - float value; - if (abnormalFlag == 0) { - value = RandomNum.getRandomFloat(ABNORMAL_MIN_FLOAT, ABNORMAL_MAX_FLOAT); - } else { - value = RandomNum.getRandomFloat(MIN_FLOAT, MAX_FLOAT); - } - sql = String.format(insertDataSql, path, sensor, time, value); - } else if (type.equals("TEXT")) { - String value; - value = RandomNum.getRandomText(STRING_LENGTH); - sql = String.format(insertDataSql, path, sensor, time, "\"" + value + "\""); - } - - statement.addBatch(sql); - statement1.addBatch(sql); - sqlCount++; - if (sqlCount >= BATCH_SQL) { - statement.executeBatch(); - statement.clearBatch(); - statement1.executeBatch(); - statement1.clearBatch(); - sqlCount = 0; - } - } - - if (abnormalFlag == 0) { - abnormalCount += 1; - } - if (abnormalCount >= ABNORMAL_LENGTH) { - abnormalCount = 0; - abnormalFlag = 1; - } - } - statement.executeBatch(); - statement.clearBatch(); - statement1.executeBatch(); - statement1.clearBatch(); - } - - /** - * main function. - * - * @param args arguments - * @throws Exception Exception - */ - public static void main(String[] args) throws Exception { - - Statement statement = null; - Statement statement1 = null; - - String path = - new File(System.getProperty(IoTDBConstant.IOTDB_HOME, null)).getParent() + File.separator - + "src" - + File.separator + "test" + File.separator + "resources" + File.separator - + "CreateTimeseries3.txt"; - Map<String, String> timeseriesMap = generateTimeseriesMapFromFile(path); - - List<String> storageGroupList = new ArrayList<>(); - storageGroupList.add("root.vehicle_history2"); - storageGroupList.add("root.vehicle_alarm2"); - storageGroupList.add("root.vehicle_temp2"); - storageGroupList.add("root.range_event2"); - - try (Connection connection1 = DriverManager - .getConnection("jdbc:iotdb://192.168.130.17:6667/", "root", "root")) { - Class.forName("org.apache.iotdb.jdbc.IoTDBDriver"); - try (Connection connection = DriverManager - .getConnection("jdbc:iotdb://localhost:6667/", "root", "root")) { - statement = connection.createStatement(); - - statement1 = connection1.createStatement(); - - setStorageGroup(statement, statement1, storageGroupList); - logger.debug("Finish set storage group."); - createTimeseries(statement, statement1, timeseriesMap); - logger.debug("Finish create timeseries."); - while (true) { - randomInsertData(statement, statement1, timeseriesMap); - } - } - } catch (Exception e) { - logger.error("", e); - } finally { - if (statement != null) { - statement.close(); - } - if (statement1 != null) { - statement1.close(); - } - } - } -} \ No newline at end of file diff --git a/server/src/test/java/org/apache/iotdb/db/sync/test/Utils.java b/server/src/test/java/org/apache/iotdb/db/sync/test/Utils.java deleted file mode 100644 index bd24a10..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/test/Utils.java +++ /dev/null @@ -1,44 +0,0 @@ -/** - * 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.iotdb.db.sync.test; - -public class Utils { - - public static String getType(String properties) { - return properties.split(",")[0]; - } - - public static String getEncode(String properties) { - return properties.split(",")[1]; - } - - private Utils() { - throw new IllegalStateException("Utility class"); - } - - public static String getPath(String timeseries) { - int lastPointIndex = timeseries.lastIndexOf('.'); - return timeseries.substring(0, lastPointIndex); - } - - public static String getSensor(String timeseries) { - int lastPointIndex = timeseries.lastIndexOf('.'); - return timeseries.substring(lastPointIndex + 1); - } -}
