You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "JackieTien97 (via GitHub)" <gi...@apache.org> on 2023/02/21 07:58:21 UTC
[GitHub] [iotdb] JackieTien97 commented on a diff in pull request #9104: [IOTDB-5454] Support shuffle function of DataExchangeModule
JackieTien97 commented on code in PR #9104:
URL: https://github.com/apache/iotdb/pull/9104#discussion_r1112478663
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/ISinkChannel.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.mpp.execution.exchange.sink;
+
+public interface ISinkChannel {
Review Comment:
Add some java doc for this class to explain the difference from ISinkHandle and its own functionality.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/ISinkChannel.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.mpp.execution.exchange.sink;
+
+public interface ISinkChannel {
Review Comment:
We should rethink about this interface. IMO, `ISinkChannel` may need to extends `ISink`, `ISinkHandle` also needs to extend `ISink`. In `ISink`, it contains all the common methods that `ISinkChannel` and `ISinkHandle` share. And then `ShuffleSinkHandle` only needs to implements `ISinkHandle`, previous `SinkHandle` and `LocalSinkHandle` can be renamed to `SinkChannel` and `LocalSinkChannel`, they only need to implement `ISinkChannel`.
And in `ShuffleSinkHandle`, it needs `List<ISinkChannel>` instead of `List<ISinkHandle>`
Some methods in previous `ISinkHandle` can be moved to `ISinChannel`, like `isFinished`.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/ISinkHandle.java:
##########
@@ -16,65 +16,75 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.execution.exchange;
+package org.apache.iotdb.db.mpp.execution.exchange.sink;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
-import java.util.List;
-
public interface ISinkHandle {
/** Get the local fragment instance ID that this sink handle belongs to. */
TFragmentInstanceId getLocalFragmentInstanceId();
- /** Get the total amount of memory used by buffered tsblocks. */
+ /** Get the total amount of memory used by buffered TsBlocks. */
long getBufferRetainedSizeInBytes();
+ default ISinkHandle getChannel(int index) {
Review Comment:
```suggestion
default ISinkChannel getChannel(int index) {
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/ISinkHandle.java:
##########
@@ -16,65 +16,75 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.execution.exchange;
+package org.apache.iotdb.db.mpp.execution.exchange.sink;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
-import java.util.List;
-
public interface ISinkHandle {
/** Get the local fragment instance ID that this sink handle belongs to. */
TFragmentInstanceId getLocalFragmentInstanceId();
- /** Get the total amount of memory used by buffered tsblocks. */
+ /** Get the total amount of memory used by buffered TsBlocks. */
long getBufferRetainedSizeInBytes();
+ default ISinkHandle getChannel(int index) {
+ throw new UnsupportedOperationException();
+ }
+
/** Get a future that will be completed when the output buffer is not full. */
ListenableFuture<?> isFull();
/**
- * Send a list of tsblocks to an unpartitioned output buffer. If no-more-tsblocks has been set,
+ * Send a {@link TsBlock} to an un-partitioned output buffer. If no-more-TsBlocks has been set,
* the invocation will be ignored. This can happen with limit queries. A {@link RuntimeException}
* will be thrown if any exception happened during the data transmission.
*/
void send(TsBlock tsBlock);
/**
- * Send a {@link TsBlock} to a specific partition. If no-more-tsblocks has been set, the send
- * tsblock call is ignored. This can happen with limit queries. A {@link RuntimeException} will be
- * thrown if any exception happened * during the data transmission.
+ * Notify the handle that there are no more TsBlocks. Any future calls to send a TsBlock should be
+ * ignored.
*/
- void send(int partition, List<TsBlock> tsBlocks);
+ void setNoMoreTsBlocks();
/**
- * Notify the handle that there are no more tsblocks. Any future calls to send a tsblock should be
- * ignored.
+ * Notify the handle that there are no more TsBlocks for the specified channel. Any future calls
+ * to send a TsBlock to the specified channel should be ignored.
+ *
+ * @param channelIndex index of the channel that should be closed
*/
- void setNoMoreTsBlocks();
+ default void setNoMoreTsBlocksOfOneChannel(int channelIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ /** Open specified channel of ISinkHandle. */
+ default void tryOpenChannel(int channelIndex) {
+ throw new UnsupportedOperationException();
+ }
Review Comment:
After we redesign the interface, these two methods don't need to be default method.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/LocalSinkHandle.java:
##########
@@ -29,15 +30,14 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.List;
import java.util.Optional;
import static com.google.common.util.concurrent.Futures.nonCancellationPropagating;
import static org.apache.iotdb.db.mpp.metric.DataExchangeCostMetricSet.SINK_HANDLE_SEND_TSBLOCK_LOCAL;
-public class LocalSinkHandle implements ISinkHandle {
+public class LocalSinkHandle implements ISinkHandle, ISinkChannel {
Review Comment:
```suggestion
public class LocalSinkChannel implements ISinkChannel {
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/LocalSinkHandle.java:
##########
@@ -47,6 +47,8 @@ public class LocalSinkHandle implements ISinkHandle {
private boolean aborted = false;
private boolean closed = false;
+ private boolean noMoreTsBlocks = false;
Review Comment:
no need to add this field, we can directly use `noMoreTsBlocks` in `SharedTsBlockQueue`
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/LocalSinkHandle.java:
##########
@@ -207,4 +205,26 @@ public void setMaxBytesCanReserve(long maxBytesCanReserve) {
// do nothing, the maxBytesCanReserve of SharedTsBlockQueue should be set by corresponding
// LocalSourceHandle
}
+
+ // region ============ ISinkChannel related ============
+
+ @Override
+ public void open() {}
+
+ @Override
+ public boolean isNoMoreTsBlocks() {
+ return noMoreTsBlocks;
+ }
+
+ @Override
+ public long getRetainedSizeInBytes() {
+ return queue.getBufferRetainedSizeInBytes();
Review Comment:
```suggestion
synchronized (queue) {
return queue.getBufferRetainedSizeInBytes();
}
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/LocalSinkHandle.java:
##########
@@ -207,4 +205,26 @@ public void setMaxBytesCanReserve(long maxBytesCanReserve) {
// do nothing, the maxBytesCanReserve of SharedTsBlockQueue should be set by corresponding
// LocalSourceHandle
}
+
+ // region ============ ISinkChannel related ============
+
+ @Override
+ public void open() {}
+
+ @Override
+ public boolean isNoMoreTsBlocks() {
+ return noMoreTsBlocks;
+ }
+
+ @Override
+ public long getRetainedSizeInBytes() {
+ return queue.getBufferRetainedSizeInBytes();
+ }
+
+ @Override
+ public int getNumOfBufferedTsBlocks() {
+ return queue.getNumOfBufferedTsBlocks();
Review Comment:
```suggestion
synchronized (queue) {
return queue.getNumOfBufferedTsBlocks();
}
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java:
##########
@@ -419,20 +442,15 @@ public MPPDataExchangeServiceImpl getOrCreateMPPDataExchangeServiceImpl() {
return mppDataExchangeService;
}
- @Override
public synchronized ISinkHandle createLocalSinkHandleForFragment(
Review Comment:
```suggestion
public synchronized ISinkChannel createLocalSinkChannelForFragment(
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java:
##########
@@ -468,19 +483,18 @@ public synchronized ISinkHandle createLocalSinkHandleForFragment(
*/
public ISinkHandle createLocalSinkHandleForPipeline(
Review Comment:
```suggestion
public ISinkChannel createLocalSinkChannelForPipeline(
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java:
##########
@@ -490,30 +504,84 @@ public ISinkHandle createSinkHandle(
// TODO: replace with callbacks to decouple MPPDataExchangeManager from
// FragmentInstanceContext
FragmentInstanceContext instanceContext) {
- if (sinkHandles.containsKey(localFragmentInstanceId)) {
- throw new IllegalStateException("Sink handle for " + localFragmentInstanceId + " exists.");
- }
- logger.debug(
+ LOGGER.debug(
"Create sink handle to plan node {} of {} for {}",
remotePlanNodeId,
remoteFragmentInstanceId,
localFragmentInstanceId);
- SinkHandle sinkHandle =
- new SinkHandle(
- remoteEndpoint,
- remoteFragmentInstanceId,
- remotePlanNodeId,
- localPlanNodeId,
+ return new SinkHandle(
+ remoteEndpoint,
+ remoteFragmentInstanceId,
+ remotePlanNodeId,
+ localPlanNodeId,
+ localFragmentInstanceId,
+ localMemoryManager,
+ executorService,
+ tsBlockSerdeFactory.get(),
+ new SinkHandleListenerImpl(instanceContext, instanceContext::failed),
+ mppDataExchangeServiceClientManager);
+ }
+
+ @Override
+ public ISinkHandle createShuffleSinkHandle(
+ List<DownStreamChannelLocation> downStreamChannelLocationList,
+ DownStreamChannelIndex downStreamChannelIndex,
+ ShuffleSinkHandle.ShuffleStrategyEnum shuffleStrategyEnum,
+ TFragmentInstanceId localFragmentInstanceId,
+ String localPlanNodeId,
+ // TODO: replace with callbacks to decouple MPPDataExchangeManager from
+ // FragmentInstanceContext
+ FragmentInstanceContext instanceContext) {
+ if (shuffleSinkHandles.containsKey(localFragmentInstanceId)) {
+ throw new IllegalStateException(
+ "ShuffleSinkHandle for " + localFragmentInstanceId + " exists.");
+ }
+
+ List<ISinkHandle> downStreamChannelList =
+ downStreamChannelLocationList.stream()
+ .map(
+ downStreamChannelLocation ->
+ createHandleForShuffleSink(
+ localFragmentInstanceId,
+ localPlanNodeId,
+ downStreamChannelLocation,
+ instanceContext))
+ .collect(Collectors.toList());
+
+ ShuffleSinkHandle shuffleSinkHandle =
+ new ShuffleSinkHandle(
localFragmentInstanceId,
- localMemoryManager,
- executorService,
- tsBlockSerdeFactory.get(),
- new SinkHandleListenerImpl(instanceContext, instanceContext::failed),
- mppDataExchangeServiceClientManager);
- sinkHandles.put(localFragmentInstanceId, sinkHandle);
- return sinkHandle;
+ downStreamChannelList,
+ downStreamChannelIndex,
+ shuffleStrategyEnum,
+ localPlanNodeId,
+ new ShuffleSinkHandleListenerImpl(instanceContext, instanceContext::failed));
+ shuffleSinkHandles.put(localFragmentInstanceId, shuffleSinkHandle);
+ return shuffleSinkHandle;
+ }
+
+ private ISinkHandle createHandleForShuffleSink(
Review Comment:
```suggestion
private ISinkChannel createChannelForShuffleSink(
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java:
##########
@@ -490,30 +504,84 @@ public ISinkHandle createSinkHandle(
// TODO: replace with callbacks to decouple MPPDataExchangeManager from
// FragmentInstanceContext
FragmentInstanceContext instanceContext) {
- if (sinkHandles.containsKey(localFragmentInstanceId)) {
- throw new IllegalStateException("Sink handle for " + localFragmentInstanceId + " exists.");
- }
- logger.debug(
+ LOGGER.debug(
"Create sink handle to plan node {} of {} for {}",
remotePlanNodeId,
remoteFragmentInstanceId,
localFragmentInstanceId);
- SinkHandle sinkHandle =
- new SinkHandle(
- remoteEndpoint,
- remoteFragmentInstanceId,
- remotePlanNodeId,
- localPlanNodeId,
+ return new SinkHandle(
+ remoteEndpoint,
+ remoteFragmentInstanceId,
+ remotePlanNodeId,
+ localPlanNodeId,
+ localFragmentInstanceId,
+ localMemoryManager,
+ executorService,
+ tsBlockSerdeFactory.get(),
+ new SinkHandleListenerImpl(instanceContext, instanceContext::failed),
+ mppDataExchangeServiceClientManager);
+ }
+
+ @Override
+ public ISinkHandle createShuffleSinkHandle(
+ List<DownStreamChannelLocation> downStreamChannelLocationList,
+ DownStreamChannelIndex downStreamChannelIndex,
+ ShuffleSinkHandle.ShuffleStrategyEnum shuffleStrategyEnum,
+ TFragmentInstanceId localFragmentInstanceId,
+ String localPlanNodeId,
+ // TODO: replace with callbacks to decouple MPPDataExchangeManager from
+ // FragmentInstanceContext
+ FragmentInstanceContext instanceContext) {
+ if (shuffleSinkHandles.containsKey(localFragmentInstanceId)) {
+ throw new IllegalStateException(
+ "ShuffleSinkHandle for " + localFragmentInstanceId + " exists.");
+ }
+
+ List<ISinkHandle> downStreamChannelList =
Review Comment:
```suggestion
List<ISinkChannel> downStreamChannelList =
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/IMPPDataExchangeManager.java:
##########
@@ -20,36 +20,35 @@
package org.apache.iotdb.db.mpp.execution.exchange;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.db.mpp.execution.driver.DriverContext;
+import org.apache.iotdb.db.mpp.execution.exchange.sink.DownStreamChannelIndex;
+import org.apache.iotdb.db.mpp.execution.exchange.sink.DownStreamChannelLocation;
+import org.apache.iotdb.db.mpp.execution.exchange.sink.ISinkHandle;
+import org.apache.iotdb.db.mpp.execution.exchange.sink.ShuffleSinkHandle;
+import org.apache.iotdb.db.mpp.execution.exchange.source.ISourceHandle;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
+import java.util.List;
+
public interface IMPPDataExchangeManager {
/**
* Create a sink handle who sends data blocks to a remote downstream fragment instance in async
* manner.
*
* @param localFragmentInstanceId ID of the local fragment instance who generates and sends data
* blocks to the sink handle.
- * @param remoteEndpoint Hostname and Port of the remote fragment instance where the data blocks
- * should be sent to.
- * @param remotePlanNodeId The sink plan node ID of the remote fragment instance.
- * @param remotePlanNodeId The plan node ID of the local fragment instance.
* @param instanceContext The context of local fragment instance.
*/
- ISinkHandle createSinkHandle(
+ ISinkHandle createShuffleSinkHandle(
+ List<DownStreamChannelLocation> downStreamChannelLocationList,
+ DownStreamChannelIndex downStreamChannelIndex,
+ ShuffleSinkHandle.ShuffleStrategyEnum shuffleStrategyEnum,
TFragmentInstanceId localFragmentInstanceId,
- TEndPoint remoteEndpoint,
- TFragmentInstanceId remoteFragmentInstanceId,
- String remotePlanNodeId,
String localPlanNodeId,
FragmentInstanceContext instanceContext);
- ISinkHandle createLocalSinkHandleForFragment(
- TFragmentInstanceId localFragmentInstanceId,
- TFragmentInstanceId remoteFragmentInstanceId,
- String remotePlanNodeId,
- FragmentInstanceContext instanceContext);
-
+ ISinkHandle createLocalSinkHandleForPipeline(DriverContext driverContext, String planNodeId);
Review Comment:
```suggestion
ISinkChannel createLocalSinkChannelForPipeline(DriverContext driverContext, String planNodeId);
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java:
##########
@@ -490,30 +504,84 @@ public ISinkHandle createSinkHandle(
// TODO: replace with callbacks to decouple MPPDataExchangeManager from
// FragmentInstanceContext
FragmentInstanceContext instanceContext) {
- if (sinkHandles.containsKey(localFragmentInstanceId)) {
- throw new IllegalStateException("Sink handle for " + localFragmentInstanceId + " exists.");
- }
- logger.debug(
+ LOGGER.debug(
"Create sink handle to plan node {} of {} for {}",
remotePlanNodeId,
remoteFragmentInstanceId,
localFragmentInstanceId);
- SinkHandle sinkHandle =
- new SinkHandle(
- remoteEndpoint,
- remoteFragmentInstanceId,
- remotePlanNodeId,
- localPlanNodeId,
+ return new SinkHandle(
+ remoteEndpoint,
+ remoteFragmentInstanceId,
+ remotePlanNodeId,
+ localPlanNodeId,
+ localFragmentInstanceId,
+ localMemoryManager,
+ executorService,
+ tsBlockSerdeFactory.get(),
+ new SinkHandleListenerImpl(instanceContext, instanceContext::failed),
+ mppDataExchangeServiceClientManager);
+ }
+
+ @Override
+ public ISinkHandle createShuffleSinkHandle(
+ List<DownStreamChannelLocation> downStreamChannelLocationList,
+ DownStreamChannelIndex downStreamChannelIndex,
+ ShuffleSinkHandle.ShuffleStrategyEnum shuffleStrategyEnum,
+ TFragmentInstanceId localFragmentInstanceId,
+ String localPlanNodeId,
+ // TODO: replace with callbacks to decouple MPPDataExchangeManager from
+ // FragmentInstanceContext
+ FragmentInstanceContext instanceContext) {
+ if (shuffleSinkHandles.containsKey(localFragmentInstanceId)) {
+ throw new IllegalStateException(
+ "ShuffleSinkHandle for " + localFragmentInstanceId + " exists.");
+ }
+
+ List<ISinkHandle> downStreamChannelList =
+ downStreamChannelLocationList.stream()
+ .map(
+ downStreamChannelLocation ->
+ createHandleForShuffleSink(
Review Comment:
```suggestion
createChannelForShuffleSink(
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/LocalSinkHandle.java:
##########
@@ -133,16 +135,12 @@ public void send(TsBlock tsBlock) {
}
}
- @Override
- public synchronized void send(int partition, List<TsBlock> tsBlocks) {
- throw new UnsupportedOperationException();
- }
-
@Override
public void setNoMoreTsBlocks() {
synchronized (queue) {
synchronized (this) {
- logger.debug("[StartSetNoMoreTsBlocksOnLocal]");
+ LOGGER.debug("[StartSetNoMoreTsBlocksOnLocal]");
+ noMoreTsBlocks = true;
Review Comment:
```suggestion
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/LocalSinkHandle.java:
##########
@@ -207,4 +205,26 @@ public void setMaxBytesCanReserve(long maxBytesCanReserve) {
// do nothing, the maxBytesCanReserve of SharedTsBlockQueue should be set by corresponding
// LocalSourceHandle
}
+
+ // region ============ ISinkChannel related ============
+
+ @Override
+ public void open() {}
+
+ @Override
+ public boolean isNoMoreTsBlocks() {
+ return noMoreTsBlocks;
Review Comment:
```suggestion
synchronized (queue) {
return queue.hasNoMoreTsBlocks();
}
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/ShuffleSinkHandle.java:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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.mpp.execution.exchange.sink;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.mpp.execution.exchange.MPPDataExchangeManager;
+import org.apache.iotdb.db.mpp.metric.QueryMetricsManager;
+import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.commons.lang3.Validate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+import static org.apache.iotdb.db.mpp.metric.DataExchangeCostMetricSet.SINK_HANDLE_SEND_TSBLOCK_REMOTE;
+
+public class ShuffleSinkHandle implements ISinkHandle {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ShuffleSinkHandle.class);
+
+ /** Each ISinkHandle in the list matches one downStream ISourceHandle */
+ private final List<ISinkHandle> downStreamChannelList;
Review Comment:
```suggestion
private final List<ISinkChannel> downStreamChannelList;
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java:
##########
@@ -468,19 +483,18 @@ public synchronized ISinkHandle createLocalSinkHandleForFragment(
*/
public ISinkHandle createLocalSinkHandleForPipeline(
DriverContext driverContext, String planNodeId) {
- logger.debug("Create local sink handle for {}", driverContext.getDriverTaskID());
+ LOGGER.debug("Create local sink handle for {}", driverContext.getDriverTaskID());
SharedTsBlockQueue queue =
new SharedTsBlockQueue(
driverContext.getDriverTaskID().getFragmentInstanceId().toThrift(),
planNodeId,
localMemoryManager);
return new LocalSinkHandle(
queue,
- new PipelineSinkHandleListenerImpl(
+ new SinkHandleListenerImpl(
driverContext.getFragmentInstanceContext(), driverContext::failed));
}
- @Override
public ISinkHandle createSinkHandle(
Review Comment:
```suggestion
public ISinkChannel createSinkChannel(
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java:
##########
@@ -1822,36 +1829,71 @@ public Operator visitExchange(ExchangeNode node, LocalExecutionPlanContext conte
}
@Override
- public Operator visitFragmentSink(FragmentSinkNode node, LocalExecutionPlanContext context) {
- if (!isSameNode(node.getDownStreamEndpoint())) {
- context.addExchangeSumNum(1);
- }
- Operator child = node.getChild().accept(this, context);
+ public Operator visitIdentitySink(IdentitySinkNode node, LocalExecutionPlanContext context) {
+ context.addExchangeSumNum(1);
+ OperatorContext operatorContext =
+ context
+ .getDriverContext()
+ .addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ IdentitySinkOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
Review Comment:
```suggestion
```
no need to record it, it's not a time consuming operation.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/sink/ShuffleHelperOperator.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.mpp.execution.operator.sink;
+
+import org.apache.iotdb.db.mpp.execution.exchange.sink.DownStreamChannelIndex;
+import org.apache.iotdb.db.mpp.execution.exchange.sink.ISinkHandle;
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class ShuffleHelperOperator implements Operator {
+ private final OperatorContext operatorContext;
+ private final List<Operator> children;
+
+ private final DownStreamChannelIndex downStreamChannelIndex;
+
+ private final ISinkHandle sinkHandle;
+
+ private final Set<Integer> unfinishedChildren;
+
+ private boolean needToReturnNull = false;
+
+ public ShuffleHelperOperator(
+ OperatorContext operatorContext,
+ List<Operator> children,
+ DownStreamChannelIndex downStreamChannelIndex,
+ ISinkHandle sinkHandle) {
+ this.operatorContext = operatorContext;
+ this.children = children;
+ this.downStreamChannelIndex = downStreamChannelIndex;
+ this.sinkHandle = sinkHandle;
+ this.unfinishedChildren = new HashSet<>(children.size());
+ for (int i = 0; i < children.size(); i++) {
+ unfinishedChildren.add(i);
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (children.get(downStreamChannelIndex.getCurrentIndex()).hasNext()) {
+ return true;
+ }
+ int currentIndex = downStreamChannelIndex.getCurrentIndex();
+ // current channel have no more data
+ sinkHandle.setNoMoreTsBlocksOfOneChannel(downStreamChannelIndex.getCurrentIndex());
Review Comment:
```suggestion
int currentIndex = downStreamChannelIndex.getCurrentIndex();
if (children.get(currentIndex).hasNext()) {
return true;
}
// current channel have no more data
sinkHandle.setNoMoreTsBlocksOfOneChannel(currentIndex);
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/ISinkHandle.java:
##########
@@ -16,65 +16,75 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.execution.exchange;
+package org.apache.iotdb.db.mpp.execution.exchange.sink;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
-import java.util.List;
-
public interface ISinkHandle {
/** Get the local fragment instance ID that this sink handle belongs to. */
TFragmentInstanceId getLocalFragmentInstanceId();
- /** Get the total amount of memory used by buffered tsblocks. */
+ /** Get the total amount of memory used by buffered TsBlocks. */
long getBufferRetainedSizeInBytes();
+ default ISinkHandle getChannel(int index) {
Review Comment:
After we redesign the interface, this method doesn't need to be default method.
##########
server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/sink/SinkHandle.java:
##########
@@ -57,9 +57,9 @@
import static org.apache.iotdb.db.mpp.metric.DataExchangeCountMetricSet.SEND_NEW_DATA_BLOCK_NUM_CALLER;
import static org.apache.iotdb.tsfile.read.common.block.TsBlockBuilderStatus.DEFAULT_MAX_TSBLOCK_SIZE_IN_BYTES;
-public class SinkHandle implements ISinkHandle {
+public class SinkHandle implements ISinkHandle, ISinkChannel {
Review Comment:
```suggestion
public class SinkChannel implements ISinkChannel {
```
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java:
##########
@@ -1822,36 +1829,71 @@ public Operator visitExchange(ExchangeNode node, LocalExecutionPlanContext conte
}
@Override
- public Operator visitFragmentSink(FragmentSinkNode node, LocalExecutionPlanContext context) {
- if (!isSameNode(node.getDownStreamEndpoint())) {
- context.addExchangeSumNum(1);
- }
- Operator child = node.getChild().accept(this, context);
+ public Operator visitIdentitySink(IdentitySinkNode node, LocalExecutionPlanContext context) {
+ context.addExchangeSumNum(1);
Review Comment:
why we don't need the previous same node judgement?
##########
server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java:
##########
@@ -1822,36 +1829,71 @@ public Operator visitExchange(ExchangeNode node, LocalExecutionPlanContext conte
}
@Override
- public Operator visitFragmentSink(FragmentSinkNode node, LocalExecutionPlanContext context) {
- if (!isSameNode(node.getDownStreamEndpoint())) {
- context.addExchangeSumNum(1);
- }
- Operator child = node.getChild().accept(this, context);
+ public Operator visitIdentitySink(IdentitySinkNode node, LocalExecutionPlanContext context) {
+ context.addExchangeSumNum(1);
+ OperatorContext operatorContext =
+ context
+ .getDriverContext()
+ .addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ IdentitySinkOperator.class.getSimpleName());
+ context.getTimeSliceAllocator().recordExecutionWeight(operatorContext, 1);
+ List<Operator> children =
+ node.getChildren().stream()
+ .map(child -> child.accept(this, context))
+ .collect(Collectors.toList());
+ checkArgument(
+ MPP_DATA_EXCHANGE_MANAGER != null, "MPP_DATA_EXCHANGE_MANAGER should not be null");
FragmentInstanceId localInstanceId = context.getInstanceContext().getId();
- FragmentInstanceId targetInstanceId = node.getDownStreamInstanceId();
- TEndPoint downStreamEndPoint = node.getDownStreamEndpoint();
+ DownStreamChannelIndex downStreamChannelIndex = new DownStreamChannelIndex(0);
+ ISinkHandle sinkHandle =
+ MPP_DATA_EXCHANGE_MANAGER.createShuffleSinkHandle(
+ node.getDownStreamChannelLocationList(),
+ downStreamChannelIndex,
+ ShuffleSinkHandle.ShuffleStrategyEnum.PLAIN,
+ localInstanceId.toThrift(),
+ node.getPlanNodeId().getId(),
+ context.getInstanceContext());
+ sinkHandle.setMaxBytesCanReserve(context.getMaxBytesOneHandleCanReserve());
+ context.getDriverContext().setSinkHandle(sinkHandle);
+
+ return new IdentitySinkOperator(operatorContext, children, downStreamChannelIndex, sinkHandle);
+ }
+
+ @Override
+ public Operator visitShuffleSink(ShuffleSinkNode node, LocalExecutionPlanContext context) {
+ context.addExchangeSumNum(1);
Review Comment:
same as above
--
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