You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/01/12 07:45:11 UTC
[GitHub] [iotdb] Cpaulyz opened a new pull request #4786: [IOTDB-2272] implement customized sync process: receiver
Cpaulyz opened a new pull request #4786:
URL: https://github.com/apache/iotdb/pull/4786
This pr mainly contains three modules of sync receiver:
* **SQL parser**
* `START PIPESERVER`
* `STOP PIPESERVER`
* `SHOW PIPESERVER <pipeName>?`
* **ReceiverService** (aka PipeServer)
* in `ReceiverService`
* It a IService and entry for processing requests about the receiving data.
* **ReceiverManager**
* in package manager and recover
* Responsible for maintaining status information and writing to logs. It provides the ability to recover when restart IOTDB.
* There are some UT in `ReceiverManagerTest` and `ReceiverLogAnalyzerTest`
design document: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=195730287
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] yschengzi commented on a change in pull request #4786: [To new_sync][IOTDB-2272] implement customized sync process: receiver
Posted by GitBox <gi...@apache.org>.
yschengzi commented on a change in pull request #4786:
URL: https://github.com/apache/iotdb/pull/4786#discussion_r789578166
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.newsync.receiver;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.receiver.collector.Collector;
+import org.apache.iotdb.db.newsync.receiver.manager.PipeInfo;
+import org.apache.iotdb.db.newsync.receiver.manager.ReceiverManager;
+import org.apache.iotdb.db.qp.physical.sys.ShowPipeServerPlan;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.service.IService;
+import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.*;
+
+public class ReceiverService implements IService {
+ private static final Logger logger = LoggerFactory.getLogger(ReceiverService.class);
+ private final SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ private static final ReceiverManager receiverManager = ReceiverManager.getInstance();
+ private Collector collector;
+
+ /** start receiver service */
+ public boolean startPipeServer() {
+ try {
+ receiverManager.startServer();
+ // TODO: start socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
Review comment:
we usually put msg to logger, instead of e.printStackTrace()
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.newsync.receiver;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.receiver.collector.Collector;
+import org.apache.iotdb.db.newsync.receiver.manager.PipeInfo;
+import org.apache.iotdb.db.newsync.receiver.manager.ReceiverManager;
+import org.apache.iotdb.db.qp.physical.sys.ShowPipeServerPlan;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.service.IService;
+import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.*;
+
+public class ReceiverService implements IService {
+ private static final Logger logger = LoggerFactory.getLogger(ReceiverService.class);
+ private final SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ private static final ReceiverManager receiverManager = ReceiverManager.getInstance();
+ private Collector collector;
+
+ /** start receiver service */
+ public boolean startPipeServer() {
+ try {
+ receiverManager.startServer();
+ // TODO: start socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ /** stop receiver service */
+ public boolean stopPipeServer() {
+ try {
+ receiverManager.stopServer();
+ // TODO: stop socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
Review comment:
please put this to logger
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncConstant.java
##########
@@ -0,0 +1,23 @@
+/*
+ * 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.newsync.conf;
+
+public class SyncConstant {
+ public static final String RECEIVER_LOG_NAME = "sync-receiver.log";
Review comment:
receiverService.log will be a better name
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.newsync.receiver;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.receiver.collector.Collector;
+import org.apache.iotdb.db.newsync.receiver.manager.PipeInfo;
+import org.apache.iotdb.db.newsync.receiver.manager.ReceiverManager;
+import org.apache.iotdb.db.qp.physical.sys.ShowPipeServerPlan;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.service.IService;
+import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.*;
+
+public class ReceiverService implements IService {
+ private static final Logger logger = LoggerFactory.getLogger(ReceiverService.class);
+ private final SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ private static final ReceiverManager receiverManager = ReceiverManager.getInstance();
+ private Collector collector;
+
+ /** start receiver service */
+ public boolean startPipeServer() {
+ try {
+ receiverManager.startServer();
+ // TODO: start socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ /** stop receiver service */
+ public boolean stopPipeServer() {
+ try {
+ receiverManager.stopServer();
+ // TODO: stop socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ /** create and start a new pipe named pipeName */
+ public void createPipe(String pipeName, String remoteIp, long startTime) throws IOException {
+ receiverManager.createPipe(pipeName, remoteIp, startTime);
+ }
+
+ /** start an existed pipe named pipeName */
+ public void startPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.startPipe(pipeName, remoteIp);
+ }
+
+ /** stop an existed pipe named pipeName */
+ public void stopPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.stopPipe(pipeName, remoteIp);
+ }
+
+ /** drop an existed pipe named pipeName */
+ public void dropPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.dropPipe(pipeName, remoteIp);
+ }
+
+ /**
+ * query by sql SHOW PIPE
+ *
+ * @return QueryDataSet contained three columns: pipe name, status and start time
+ */
+ public QueryDataSet showPipe(ShowPipeServerPlan plan) {
+ ListDataSet dataSet =
+ new ListDataSet(
+ Arrays.asList(
+ new PartialPath(COLUMN_PIPE_NAME, false),
+ new PartialPath(COLUMN_PIPE_REMOTE_IP, false),
+ new PartialPath(COLUMN_PIPE_STATUS, false),
+ new PartialPath(COLUMN_PIPE_ROLE, false),
+ new PartialPath(COLUMN_CREATED_TIME, false)),
+ Arrays.asList(
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT));
+ List<PipeInfo> pipeInfos;
+ if (!StringUtils.isEmpty(plan.getPipeName())) {
+ pipeInfos = receiverManager.getPipeInfos(plan.getPipeName());
+ } else {
+ pipeInfos = receiverManager.getAllPipeInfos();
+ }
+ for (PipeInfo pipeInfo : pipeInfos) {
+ putPipeRecord(dataSet, pipeInfo);
+ }
+ return dataSet;
+ }
+
+ private void putPipeRecord(ListDataSet dataSet, PipeInfo pipeInfo) {
+ RowRecord rowRecord = new RowRecord(0);
+ Field pipeNameField = new Field(TSDataType.TEXT);
+ Field pipeRemoteIp = new Field(TSDataType.TEXT);
+ Field pipeStatusField = new Field(TSDataType.TEXT);
+ Field pipeRoleField = new Field(TSDataType.TEXT);
+ Field pipeCreateTimeField = new Field(TSDataType.TEXT);
+ pipeNameField.setBinaryV(new Binary(pipeInfo.getPipeName()));
+ pipeRemoteIp.setBinaryV(new Binary(pipeInfo.getRemoteIp()));
+ pipeStatusField.setBinaryV(new Binary(pipeInfo.getStatus().name()));
+ pipeRoleField.setBinaryV(new Binary("receiver"));
Review comment:
there is no "sender" role, column role can be deleted
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/recovery/ReceiverLogAnalyzer.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.newsync.receiver.recovery;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.newsync.conf.SyncConstant;
+import org.apache.iotdb.db.newsync.receiver.manager.PipeInfo;
+import org.apache.iotdb.db.newsync.receiver.manager.PipeStatus;
+import org.apache.iotdb.db.newsync.sender.conf.SenderConf;
+import org.apache.iotdb.db.service.ServiceType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.util.HashMap;
+import java.util.Map;
+
+public class ReceiverLogAnalyzer {
+ private static final Logger logger = LoggerFactory.getLogger(ReceiverLogAnalyzer.class);
+ private static boolean pipeServerEnable =
+ false; // record recovery result of receiver server status
+ private static Map<String, Map<String, PipeInfo>> pipeInfoMap =
+ new HashMap<>(); // record recovery result of receiver server status
+
+ public static void scan() throws StartupException {
+ logger.info("Start to recover all sync state for sync receiver.");
+ pipeInfoMap = new HashMap<>();
+ pipeServerEnable = false;
+ File logFile = new File(SenderConf.sysDir, SyncConstant.RECEIVER_LOG_NAME);
+ try (BufferedReader loadReader = new BufferedReader(new FileReader(logFile))) {
+ String line;
+ int lineNum = 0;
+ while ((line = loadReader.readLine()) != null) {
+ lineNum++;
+ try {
+ analyzeLog(line);
+ } catch (Exception e) {
+ logger.error("Receiver log recovery error: log file parse error at line " + lineNum);
+ e.printStackTrace();
Review comment:
please put it in logger.
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.newsync.receiver;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.receiver.collector.Collector;
+import org.apache.iotdb.db.newsync.receiver.manager.PipeInfo;
+import org.apache.iotdb.db.newsync.receiver.manager.ReceiverManager;
+import org.apache.iotdb.db.qp.physical.sys.ShowPipeServerPlan;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.service.IService;
+import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.*;
+
+public class ReceiverService implements IService {
+ private static final Logger logger = LoggerFactory.getLogger(ReceiverService.class);
+ private final SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ private static final ReceiverManager receiverManager = ReceiverManager.getInstance();
+ private Collector collector;
+
+ /** start receiver service */
+ public boolean startPipeServer() {
+ try {
+ receiverManager.startServer();
+ // TODO: start socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ /** stop receiver service */
+ public boolean stopPipeServer() {
+ try {
+ receiverManager.stopServer();
+ // TODO: stop socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ /** create and start a new pipe named pipeName */
+ public void createPipe(String pipeName, String remoteIp, long startTime) throws IOException {
+ receiverManager.createPipe(pipeName, remoteIp, startTime);
+ }
+
+ /** start an existed pipe named pipeName */
+ public void startPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.startPipe(pipeName, remoteIp);
+ }
+
+ /** stop an existed pipe named pipeName */
+ public void stopPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.stopPipe(pipeName, remoteIp);
+ }
+
+ /** drop an existed pipe named pipeName */
+ public void dropPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.dropPipe(pipeName, remoteIp);
+ }
+
+ /**
+ * query by sql SHOW PIPE
+ *
+ * @return QueryDataSet contained three columns: pipe name, status and start time
+ */
+ public QueryDataSet showPipe(ShowPipeServerPlan plan) {
+ ListDataSet dataSet =
+ new ListDataSet(
+ Arrays.asList(
+ new PartialPath(COLUMN_PIPE_NAME, false),
+ new PartialPath(COLUMN_PIPE_REMOTE_IP, false),
+ new PartialPath(COLUMN_PIPE_STATUS, false),
+ new PartialPath(COLUMN_PIPE_ROLE, false),
+ new PartialPath(COLUMN_CREATED_TIME, false)),
+ Arrays.asList(
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT));
+ List<PipeInfo> pipeInfos;
+ if (!StringUtils.isEmpty(plan.getPipeName())) {
+ pipeInfos = receiverManager.getPipeInfos(plan.getPipeName());
+ } else {
+ pipeInfos = receiverManager.getAllPipeInfos();
+ }
+ for (PipeInfo pipeInfo : pipeInfos) {
+ putPipeRecord(dataSet, pipeInfo);
+ }
+ return dataSet;
+ }
+
+ private void putPipeRecord(ListDataSet dataSet, PipeInfo pipeInfo) {
+ RowRecord rowRecord = new RowRecord(0);
+ Field pipeNameField = new Field(TSDataType.TEXT);
+ Field pipeRemoteIp = new Field(TSDataType.TEXT);
+ Field pipeStatusField = new Field(TSDataType.TEXT);
+ Field pipeRoleField = new Field(TSDataType.TEXT);
+ Field pipeCreateTimeField = new Field(TSDataType.TEXT);
+ pipeNameField.setBinaryV(new Binary(pipeInfo.getPipeName()));
+ pipeRemoteIp.setBinaryV(new Binary(pipeInfo.getRemoteIp()));
+ pipeStatusField.setBinaryV(new Binary(pipeInfo.getStatus().name()));
+ pipeRoleField.setBinaryV(new Binary("receiver"));
+ pipeCreateTimeField.setBinaryV(new Binary(dateformat.format(pipeInfo.getCreateTime())));
Review comment:
DatetimeUtils.convertLongToDate(pipeInfo.getCreateTime()) can be better
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/manager/PipeStatus.java
##########
@@ -0,0 +1,26 @@
+/*
+ * 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.newsync.receiver.manager;
+
+public enum PipeStatus {
+ RUNNING,
+ PAUSE,
+ DROP,
+ FAILED
Review comment:
There is no FAILED in sender, only DROP for FAILED situation.
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
##########
@@ -3060,4 +3060,23 @@ private void setMap(IoTDBSqlParser.AlterClauseContext ctx, Map<String, String> a
throw new SQLParserException(DELETE_RANGE_ERROR_MSG);
}
}
+
+ @Override
+ public Operator visitStartPipeServer(IoTDBSqlParser.StartPipeServerContext ctx) {
+ return new StartPipeServerOperator(SQLConstant.TOK_PIPE_SERVER_START);
+ }
+
+ @Override
+ public Operator visitStopPipeServer(IoTDBSqlParser.StopPipeServerContext ctx) {
+ return new StopPipeServerOperator(SQLConstant.TOK_PIPE_SERVER_STOP);
+ }
+
+ @Override
+ public Operator visitShowPipeServer(IoTDBSqlParser.ShowPipeServerContext ctx) {
+ if (ctx.pipeName != null) {
+ return new ShowPipeServerOperator(ctx.pipeName.getText(), SQLConstant.TOK_SHOW_PIPE_SERVER);
+ } else {
+ return new ShowPipeServerOperator(SQLConstant.TOK_SHOW_PIPE_SERVER);
+ }
+ }
Review comment:
please add this in line 2283, the code region from line 2145 to line 2283 is for sync
##########
File path: server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
##########
@@ -193,9 +193,9 @@ public PhysicalPlan generatePhysicalPlan(PhysicalGenerator generator)
PRUNE_TEMPLATE,
APPEND_TEMPLATE,
DROP_TEMPLATE,
-
+ START_PIPE_SERVER,
+ STOP_PIPE_SERVER,
Review comment:
please append this to the end.
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.newsync.receiver;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.receiver.collector.Collector;
+import org.apache.iotdb.db.newsync.receiver.manager.PipeInfo;
+import org.apache.iotdb.db.newsync.receiver.manager.ReceiverManager;
+import org.apache.iotdb.db.qp.physical.sys.ShowPipeServerPlan;
+import org.apache.iotdb.db.query.dataset.ListDataSet;
+import org.apache.iotdb.db.service.IService;
+import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.*;
+
+public class ReceiverService implements IService {
+ private static final Logger logger = LoggerFactory.getLogger(ReceiverService.class);
+ private final SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ private static final ReceiverManager receiverManager = ReceiverManager.getInstance();
+ private Collector collector;
+
+ /** start receiver service */
+ public boolean startPipeServer() {
+ try {
+ receiverManager.startServer();
+ // TODO: start socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ /** stop receiver service */
+ public boolean stopPipeServer() {
+ try {
+ receiverManager.stopServer();
+ // TODO: stop socket and collector
+ } catch (IOException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ /** create and start a new pipe named pipeName */
+ public void createPipe(String pipeName, String remoteIp, long startTime) throws IOException {
+ receiverManager.createPipe(pipeName, remoteIp, startTime);
+ }
+
+ /** start an existed pipe named pipeName */
+ public void startPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.startPipe(pipeName, remoteIp);
+ }
+
+ /** stop an existed pipe named pipeName */
+ public void stopPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.stopPipe(pipeName, remoteIp);
+ }
+
+ /** drop an existed pipe named pipeName */
+ public void dropPipe(String pipeName, String remoteIp) throws IOException {
+ receiverManager.dropPipe(pipeName, remoteIp);
+ }
+
+ /**
+ * query by sql SHOW PIPE
+ *
+ * @return QueryDataSet contained three columns: pipe name, status and start time
+ */
+ public QueryDataSet showPipe(ShowPipeServerPlan plan) {
+ ListDataSet dataSet =
+ new ListDataSet(
+ Arrays.asList(
+ new PartialPath(COLUMN_PIPE_NAME, false),
+ new PartialPath(COLUMN_PIPE_REMOTE_IP, false),
+ new PartialPath(COLUMN_PIPE_STATUS, false),
+ new PartialPath(COLUMN_PIPE_ROLE, false),
+ new PartialPath(COLUMN_CREATED_TIME, false)),
+ Arrays.asList(
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT,
+ TSDataType.TEXT));
+ List<PipeInfo> pipeInfos;
+ if (!StringUtils.isEmpty(plan.getPipeName())) {
+ pipeInfos = receiverManager.getPipeInfos(plan.getPipeName());
+ } else {
+ pipeInfos = receiverManager.getAllPipeInfos();
+ }
+ for (PipeInfo pipeInfo : pipeInfos) {
+ putPipeRecord(dataSet, pipeInfo);
+ }
+ return dataSet;
+ }
+
+ private void putPipeRecord(ListDataSet dataSet, PipeInfo pipeInfo) {
+ RowRecord rowRecord = new RowRecord(0);
+ Field pipeNameField = new Field(TSDataType.TEXT);
+ Field pipeRemoteIp = new Field(TSDataType.TEXT);
+ Field pipeStatusField = new Field(TSDataType.TEXT);
+ Field pipeRoleField = new Field(TSDataType.TEXT);
+ Field pipeCreateTimeField = new Field(TSDataType.TEXT);
+ pipeNameField.setBinaryV(new Binary(pipeInfo.getPipeName()));
+ pipeRemoteIp.setBinaryV(new Binary(pipeInfo.getRemoteIp()));
+ pipeStatusField.setBinaryV(new Binary(pipeInfo.getStatus().name()));
+ pipeRoleField.setBinaryV(new Binary("receiver"));
+ pipeCreateTimeField.setBinaryV(new Binary(dateformat.format(pipeInfo.getCreateTime())));
+ rowRecord.addField(pipeNameField);
+ rowRecord.addField(pipeRemoteIp);
+ rowRecord.addField(pipeStatusField);
+ rowRecord.addField(pipeRoleField);
+ rowRecord.addField(pipeCreateTimeField);
+ dataSet.putRecord(rowRecord);
+ }
+
+ private ReceiverService() {
+ collector = new Collector();
+ }
+
+ public static ReceiverService getInstance() {
+ return ReceiverServiceHolder.INSTANCE;
+ }
+
+ /** IService * */
+ @Override
+ public void start() throws StartupException {
+ receiverManager.init();
+ if (receiverManager.isPipeServerEnable()) {
+ startPipeServer();
+ }
+ }
+
+ @Override
+ public void stop() {
+ stopPipeServer();
+ try {
+ receiverManager.close();
+ } catch (IOException e) {
+ e.printStackTrace();
Review comment:
please put it to logger
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/load/SchemaLoader.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.newsync.receiver.load;
+
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.metadata.StorageGroupAlreadySetException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+
+import java.io.IOException;
+
+/**
+ * This loader is used to PhysicalPlan. Support four types of physical plans: CREATE_TIMESERIES |
+ * CREATE_ALIGNED_TIMESERIES | DELETE_TIMESERIES | SET_STORAGE_GROUP
+ */
+public class SchemaLoader implements ILoader {
+
+ private PhysicalPlan plan;
+
+ public SchemaLoader(PhysicalPlan plan) {
+ this.plan = plan;
+ }
+
+ @Override
+ public void load() throws IOException, MetadataException {
+ try {
+ MManager.getInstance().operation(plan);
+ } catch (StorageGroupAlreadySetException e) {
+ // do nothing
Review comment:
i think record a logger here can be better for debug:)
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/manager/ReceiverManager.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.newsync.receiver.manager;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.newsync.receiver.recovery.ReceiverLog;
+import org.apache.iotdb.db.newsync.receiver.recovery.ReceiverLogAnalyzer;
+import org.apache.iotdb.db.service.ServiceType;
+
+import java.io.IOException;
+import java.util.*;
+
+public class ReceiverManager {
+
+ private boolean pipeServerEnable;
+ // <pipeName, <remoteIp, pipeInfo>>
+ private Map<String, Map<String, PipeInfo>> pipeInfoMap;
Review comment:
a sender may has many pipes with the same pipeNames, because the pipe can be dropped, Map<remoteIP, List<pipeName, pipeInfo>> is a better choice.
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/manager/ReceiverManager.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.newsync.receiver.manager;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.newsync.receiver.recovery.ReceiverLog;
+import org.apache.iotdb.db.newsync.receiver.recovery.ReceiverLogAnalyzer;
+import org.apache.iotdb.db.service.ServiceType;
+
+import java.io.IOException;
+import java.util.*;
+
+public class ReceiverManager {
+
+ private boolean pipeServerEnable;
+ // <pipeName, <remoteIp, pipeInfo>>
+ private Map<String, Map<String, PipeInfo>> pipeInfoMap;
+ private ReceiverLog log;
+
+ public void init() throws StartupException {
+ try {
+ log = new ReceiverLog();
+ } catch (IOException e) {
+ e.printStackTrace();
Review comment:
please put this in logger.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] qiaojialin merged pull request #4786: [To new_sync][IOTDB-2272] implement customized sync process: receiver
Posted by GitBox <gi...@apache.org>.
qiaojialin merged pull request #4786:
URL: https://github.com/apache/iotdb/pull/4786
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] qiaojialin merged pull request #4786: [To new_sync][IOTDB-2272] implement customized sync process: receiver
Posted by GitBox <gi...@apache.org>.
qiaojialin merged pull request #4786:
URL: https://github.com/apache/iotdb/pull/4786
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] Cpaulyz commented on a change in pull request #4786: [To new_sync][IOTDB-2272] implement customized sync process: receiver
Posted by GitBox <gi...@apache.org>.
Cpaulyz commented on a change in pull request #4786:
URL: https://github.com/apache/iotdb/pull/4786#discussion_r790111645
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/manager/ReceiverManager.java
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.newsync.receiver.manager;
+
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.newsync.receiver.recovery.ReceiverLog;
+import org.apache.iotdb.db.newsync.receiver.recovery.ReceiverLogAnalyzer;
+import org.apache.iotdb.db.service.ServiceType;
+
+import java.io.IOException;
+import java.util.*;
+
+public class ReceiverManager {
+
+ private boolean pipeServerEnable;
+ // <pipeName, <remoteIp, pipeInfo>>
+ private Map<String, Map<String, PipeInfo>> pipeInfoMap;
Review comment:
Yes, maybe there are many pipes with the same pipeNames. For thses two choice, `Map<remoteIP, List<pipeName, pipeInfo>>` have the advantage over storage cost.
However, considering sql `SHOW PIPESERVER [pipeName] `, we need to provide function `getPipeInfos(String pipeName)`. The time complexity of the origin choice is O(1) while another is O(n).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org