This is an automated email from the ASF dual-hosted git repository.
ericpai pushed a commit to branch new_mpp
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/new_mpp by this push:
new 50d3bd9 [To_new_mpp] data block manager interface (#5280)
50d3bd9 is described below
commit 50d3bd9bd710e3827b516ed740c297fb37057f8b
Author: Zhong Wang <[email protected]>
AuthorDate: Sat Mar 19 10:11:47 2022 +0800
[To_new_mpp] data block manager interface (#5280)
* [To_new_mpp] data block manager interface
---
.../org/apache/iotdb/db/concurrent/ThreadName.java | 2 +
.../iotdb/db/mpp/buffer/DataBlockManager.java | 116 +++++++++++++++++++++
.../db/mpp/buffer/DataBlockManagerService.java | 90 ++++++++++++++++
...a => DataBlockManagerServiceThriftHandler.java} | 28 +++--
.../mpp/buffer/DataBlockServiceClientFactory.java | 44 ++++++++
.../iotdb/db/mpp/buffer/DataBlockServiceImpl.java | 50 +++++++++
.../iotdb/db/mpp/buffer/IDataBlockManager.java | 80 --------------
.../buffer/{SinkHandle.java => ISinkHandle.java} | 35 ++++---
.../apache/iotdb/db/mpp/buffer/ISourceHandle.java | 26 +++--
.../apache/iotdb/db/mpp/buffer/SourceHandle.java | 70 +++++++++++--
.../db/mpp/schedule/IFragmentInstanceManager.java | 1 -
.../org/apache/iotdb/db/service/ServiceType.java | 3 +-
.../src/main/thrift/{common.thrift => mpp.thrift} | 38 ++++++-
13 files changed, 455 insertions(+), 128 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
b/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
index 18ab028..ffce510 100644
--- a/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
+++ b/server/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
@@ -70,6 +70,8 @@ public enum ThreadName {
CLUSTER_DATA_HEARTBEAT_RPC_SERVICE("ClusterDataHeartbeatRPC"),
CLUSTER_DATA_HEARTBEAT_RPC_CLIENT("ClusterDataHeartbeatRPC-Client"),
Cluster_Monitor("ClusterMonitor"),
+ DATA_BLOCK_MANAGER_SERVICE("DataBlockManagerService"),
+ DATA_BLOCK_MANAGER_CLIENT("DataBlockManagerService-Client"),
;
private final String name;
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManager.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManager.java
new file mode 100644
index 0000000..9574aa6
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManager.java
@@ -0,0 +1,116 @@
+/*
+ * 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.buffer;
+
+import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.db.mpp.memory.LocalMemoryManager;
+import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
+
+import org.apache.commons.lang3.Validate;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+
+public class DataBlockManager {
+
+ public static class FragmentInstanceInfo {
+ private String hostname;
+ private String queryId;
+ private String fragmentId;
+ private String instanceId;
+
+ public FragmentInstanceInfo(
+ String hostname, String queryId, String fragmentId, String instanceId)
{
+ this.hostname = Validate.notNull(hostname);
+ this.queryId = Validate.notNull(queryId);
+ this.fragmentId = Validate.notNull(fragmentId);
+ this.instanceId = Validate.notNull(instanceId);
+ }
+
+ public String getHostname() {
+ return hostname;
+ }
+
+ public String getQueryId() {
+ return queryId;
+ }
+
+ public String getFragmentId() {
+ return fragmentId;
+ }
+
+ public String getInstanceId() {
+ return instanceId;
+ }
+ }
+
+ /**
+ * Create a sink handle.
+ *
+ * @param local The {@link FragmentInstanceInfo} of local fragment instance.
+ * @param remote The {@link FragmentInstanceInfo} of downstream instance.
+ */
+ public ISinkHandle createSinkHandle(FragmentInstanceInfo local,
FragmentInstanceInfo remote) {
+ throw new UnsupportedOperationException();
+ }
+
+ public ISinkHandle createPartitionedSinkHandle(
+ FragmentInstanceInfo local, List<FragmentInstanceInfo> remotes) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Create a source handle.
+ *
+ * @param local The {@link FragmentInstanceInfo} of local fragment instance.
+ * @param remote The {@link FragmentInstanceInfo} of downstream instance.
+ */
+ public ISourceHandle createSourceHandle(FragmentInstanceInfo local,
FragmentInstanceInfo remote) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Release all the related resources, including data blocks that are not yet
sent to downstream
+ * fragment instances.
+ *
+ * <p>This method should be called when a fragment instance finished in an
abnormal state.
+ */
+ void forceDeregisterFragmentInstance(FragmentInstanceTask task) {
+ throw new UnsupportedOperationException();
+ }
+
+ LocalMemoryManager localMemoryManager;
+ ScheduledExecutorService scheduledExecutorService;
+ DataBlockServiceClientFactory clientFactory;
+ Map<String, Map<String, Map<String, ISourceHandle>>> sourceHandles;
+ Map<String, Map<String, Map<String, ISinkHandle>>> sinkHandles;
+
+ public DataBlockManager(LocalMemoryManager localMemoryManager) {
+ this.localMemoryManager = Validate.notNull(localMemoryManager);
+ // TODO: configurable number of threads
+ scheduledExecutorService =
+ IoTDBThreadPoolFactory.newScheduledThreadPoolWithDaemon(5,
"get-data-block");
+ clientFactory = new DataBlockServiceClientFactory();
+ sourceHandles = new HashMap<>();
+ sinkHandles = new HashMap<>();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManagerService.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManagerService.java
new file mode 100644
index 0000000..76048a0
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManagerService.java
@@ -0,0 +1,90 @@
+/*
+ * 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.buffer;
+
+import org.apache.iotdb.db.concurrent.ThreadName;
+import org.apache.iotdb.db.exception.runtime.RPCServiceException;
+import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.db.service.thrift.ThriftService;
+import org.apache.iotdb.db.service.thrift.ThriftServiceThread;
+import org.apache.iotdb.mpp.rpc.thrift.DataBlockService.Processor;
+
+public class DataBlockManagerService extends ThriftService {
+
+ private DataBlockServiceImpl impl;
+
+ @Override
+ public ThriftService getImplementation() {
+ return DataBlockManagerServiceHolder.INSTANCE;
+ }
+
+ @Override
+ public void initTProcessor()
+ throws ClassNotFoundException, IllegalAccessException,
InstantiationException {
+ impl = new DataBlockServiceImpl();
+ processor = new Processor<>(impl);
+ }
+
+ @Override
+ public void initThriftServiceThread()
+ throws IllegalAccessException, InstantiationException,
ClassNotFoundException {
+ try {
+ thriftServiceThread =
+ new ThriftServiceThread(
+ processor,
+ getID().getName(),
+ ThreadName.DATA_BLOCK_MANAGER_CLIENT.getName(),
+ getBindIP(),
+ getBindPort(),
+ // TODO: hard coded maxWorkerThreads & timeoutSecond
+ 32,
+ 60,
+ new DataBlockManagerServiceThriftHandler(),
+ // TODO: hard coded compress strategy
+ true);
+ } catch (RPCServiceException e) {
+ throw new IllegalAccessException(e.getMessage());
+ }
+
thriftServiceThread.setName(ThreadName.DATA_BLOCK_MANAGER_SERVICE.getName());
+ }
+
+ @Override
+ public String getBindIP() {
+ // TODO: hard coded bind IP.
+ return "0.0.0.0";
+ }
+
+ @Override
+ public int getBindPort() {
+ // TODO: hard coded bind port.
+ return 7777;
+ }
+
+ @Override
+ public ServiceType getID() {
+ return ServiceType.DATA_BLOCK_MANAGER_SERVICE;
+ }
+
+ private static class DataBlockManagerServiceHolder {
+ private static final DataBlockManagerService INSTANCE = new
DataBlockManagerService();
+
+ private DataBlockManagerServiceHolder() {}
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManagerServiceThriftHandler.java
similarity index 53%
copy from server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
copy to
server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManagerServiceThriftHandler.java
index 260c80e..d7d40b8 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockManagerServiceThriftHandler.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -16,21 +16,29 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.buffer;
-import com.google.common.util.concurrent.ListenableFuture;
+package org.apache.iotdb.db.mpp.buffer;
-import java.io.Closeable;
-import java.nio.ByteBuffer;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.server.ServerContext;
+import org.apache.thrift.server.TServerEventHandler;
+import org.apache.thrift.transport.TTransport;
-public interface SourceHandle extends Closeable {
+public class DataBlockManagerServiceThriftHandler implements
TServerEventHandler {
- ByteBuffer receive();
+ @Override
+ public void preServe() {}
- boolean isFinished();
+ @Override
+ public ServerContext createContext(TProtocol tProtocol, TProtocol
tProtocol1) {
+ return null;
+ }
- ListenableFuture<Void> isBlocked();
+ @Override
+ public void deleteContext(
+ ServerContext serverContext, TProtocol tProtocol, TProtocol tProtocol1)
{}
@Override
- void close();
+ public void processContext(
+ ServerContext serverContext, TTransport tTransport, TTransport
tTransport1) {}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockServiceClientFactory.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockServiceClientFactory.java
new file mode 100644
index 0000000..4f4765c
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockServiceClientFactory.java
@@ -0,0 +1,44 @@
+/*
+ * 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.buffer;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.mpp.rpc.thrift.DataBlockService;
+import org.apache.iotdb.mpp.rpc.thrift.DataBlockService.Client;
+import org.apache.iotdb.rpc.RpcTransportFactory;
+
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+
+public class DataBlockServiceClientFactory {
+ public DataBlockService.Client getDataBlockServiceClient(String hostname,
int port)
+ throws TTransportException {
+ TTransport transport =
RpcTransportFactory.INSTANCE.getTransportWithNoTimeout(hostname, port);
+ transport.open();
+ TProtocol protocol =
+
IoTDBDescriptor.getInstance().getConfig().isRpcThriftCompressionEnable()
+ ? new TCompactProtocol(transport)
+ : new TBinaryProtocol(transport);
+ return new Client(protocol);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockServiceImpl.java
new file mode 100644
index 0000000..72058cb
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/DataBlockServiceImpl.java
@@ -0,0 +1,50 @@
+/*
+ * 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.buffer;
+
+import org.apache.iotdb.mpp.rpc.thrift.DataBlockService;
+import org.apache.iotdb.mpp.rpc.thrift.EndOfDataBlockEvent;
+import org.apache.iotdb.mpp.rpc.thrift.GetDataBlockReqest;
+import org.apache.iotdb.mpp.rpc.thrift.GetDataBlockResponse;
+import org.apache.iotdb.mpp.rpc.thrift.NewDataBlockEvent;
+
+import org.apache.thrift.TException;
+
+public class DataBlockServiceImpl implements DataBlockService.Iface {
+
+ public DataBlockServiceImpl() {
+ super();
+ }
+
+ @Override
+ public GetDataBlockResponse getDataBlock(GetDataBlockReqest req) throws
TException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void onNewDataBlockEvent(NewDataBlockEvent e) throws TException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void onEndOfDataBlockEvent(EndOfDataBlockEvent e) throws TException {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/IDataBlockManager.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/IDataBlockManager.java
deleted file mode 100644
index 391db95..0000000
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/IDataBlockManager.java
+++ /dev/null
@@ -1,80 +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.mpp.buffer;
-
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceID;
-import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceTask;
-import org.apache.iotdb.mpp.common.ITSBlock;
-
-public interface IDataBlockManager {
-
- /**
- * Register a new fragment instance. The block manager will start looking
for upstream data blocks
- * and flushing data blocks generated to downstream fragment instances.
- */
- void registerFragmentInstance(FragmentInstanceTask task);
-
- /**
- * Deregister a fragment instance. The block manager will stop looking for
upstream data blocks
- * and release the input data blocks, but will keep flushing data blocks to
downstream fragment
- * instances until all the data blocks are sent. Once all the data blocks
are sent, the output
- * data blocks will be release.
- *
- * <p>This method should be called when a fragment instance finished in a
normal state.
- */
- void deregisterFragmentInstance(FragmentInstanceTask task);
-
- /**
- * Deregister a fragment instance. The block manager will release all the
related resources.
- * Including data blocks that are not yet sent to downstream fragment
instances.
- *
- * <p>This method should be called when a fragment instance finished in an
abnormal state.
- */
- void forceDeregisterFragmentInstance(FragmentInstanceTask task);
-
- /**
- * Put a data block to the output buffer for downstream fragment instances.
Will throw an {@link
- * IllegalStateException} if the output buffer is full.
- *
- * <p>Once the block be put into the output buffer, the data block manager
will notify downstream
- * fragment instances that a new data block is available.
- *
- * @param instanceID ID of fragment instance that generates the block.
- * @return If there are enough memory for the next block.
- */
- boolean putDataBlock(FragmentInstanceID instanceID, ITSBlock block);
-
- /**
- * Check if there are data blocks from the specified upstream fragment
instance.
- *
- * @param instanceID ID of the upstream fragment instance.
- * @return If there are available data blocks.
- */
- boolean hasDataBlock(FragmentInstanceID instanceID);
-
- /**
- * Get a data block from the input buffer of specified upstream fragment
instance. Will throw an
- * {@link IllegalStateException} if the input buffer is empty.
- *
- * @param instanceID ID of the upstream fragment instance.
- * @return A data block.
- */
- ITSBlock getDataBlock(FragmentInstanceID instanceID);
-}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SinkHandle.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISinkHandle.java
similarity index 53%
rename from server/src/main/java/org/apache/iotdb/db/mpp/buffer/SinkHandle.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISinkHandle.java
index 0c78196..533cb40 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SinkHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISinkHandle.java
@@ -18,39 +18,40 @@
*/
package org.apache.iotdb.db.mpp.buffer;
-import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.iotdb.db.mpp.common.TsBlock;
-import java.nio.ByteBuffer;
+import com.google.common.util.concurrent.ListenableFuture;
-public interface SinkHandle {
+public interface ISinkHandle extends AutoCloseable {
- /** Get a future that will be completed when the buffer is not full. */
+ /** Get a future that will be completed when the output buffer is not full.
*/
ListenableFuture<Void> isFull();
/**
- * Sends a tsBlock to an unpartitioned buffer. If no-more-tsBlocks has been
set, the send tsBlock
- * call is ignored. This can happen with limit queries.
+ * Send a {@link TsBlock} to an unpartitioned output buffer. If
no-more-tsblocks has been set, the
+ * send tsblock call is ignored. This can happen with limit queries.
*/
- void send(ByteBuffer tsBlock);
+ void send(TsBlock tsBlock);
/**
- * Sends a tsBlock to a specific partition. If no-more-tsBlocks has been
set, the send tsBlock
- * call is ignored. This can happen with limit queries.
+ * 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.
*/
- void send(int partition, ByteBuffer tsBlock);
+ void send(int partition, TsBlock tsBlock);
/**
- * Notify SinkHandle that no more tsBlocks will be sent. Any future calls to
send a tsBlock are
- * ignored.
+ * Notify the handle that no more tsblocks will be sent. Any future calls to
send a tsblock should
+ * be ignored.
*/
void setNoMoreTsBlocks();
- /** close the sink handle, discarding all tsBlocks which may still in memory
buffer. */
- void close();
-
/**
- * Abort the sink handle, discarding all tsBlocks which may still in memory
buffer, but blocking
- * readers. It is expected that readers will be unblocked when the failed
query is cleaned up.
+ * Close the handle. Keep the output buffer until all tsblocks are fetched
by downstream
+ * instances.
*/
+ @Override
+ void close();
+
+ /** Abort the sink handle, discarding all tsblocks which may still be in
memory buffer. */
void abort();
}
diff --git a/thrift/src/main/thrift/common.thrift
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISourceHandle.java
similarity index 53%
copy from thrift/src/main/thrift/common.thrift
copy to server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISourceHandle.java
index 967deed..b33d3ef 100644
--- a/thrift/src/main/thrift/common.thrift
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISourceHandle.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -16,12 +16,26 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.iotdb.db.mpp.buffer;
-namespace java org.apache.iotdb.mpp.common.rpc.thrift
+import org.apache.iotdb.db.mpp.common.TsBlock;
+import com.google.common.util.concurrent.ListenableFuture;
-struct FragmentInstanceID {
- 1: required string queryID
- 2: required string fragmentID
- 3: required string instanceID
+import java.io.Closeable;
+
+public interface ISourceHandle extends Closeable {
+
+ /** Get a {@link TsBlock} from the input buffer. */
+ TsBlock receive();
+
+ /** Check if there are more tsblocks. */
+ boolean isFinished();
+
+ /** Get a future that will be completed when the input buffer is not empty.
*/
+ ListenableFuture<Void> isBlocked();
+
+ /** Close the handle. Discarding all tsblocks which may still be in memory
buffer. */
+ @Override
+ void close();
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
index 260c80e..b1373d7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -16,21 +16,75 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.mpp.buffer;
+import org.apache.iotdb.db.mpp.common.TsBlock;
+
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.commons.lang3.Validate;
+
+import java.util.ArrayDeque;
+import java.util.Queue;
+
+import static
com.google.common.util.concurrent.Futures.nonCancellationPropagating;
+
+public class SourceHandle implements ISourceHandle {
-import java.io.Closeable;
-import java.nio.ByteBuffer;
+ private final long bufferCapacityInBytes;
-public interface SourceHandle extends Closeable {
+ private final Queue<TsBlock> bufferedTsBlocks = new ArrayDeque<>();
+ private volatile SettableFuture<Void> blocked = SettableFuture.create();
+ private volatile long bufferRetainedSizeInBytes;
+ private boolean finished;
+ private boolean closed;
+ private Throwable throwable;
- ByteBuffer receive();
+ public SourceHandle(long bufferCapacityInBytes) {
+ Validate.isTrue(bufferCapacityInBytes > 0L, "capacity cannot be less or
equal to zero.");
+ this.bufferCapacityInBytes = bufferCapacityInBytes;
+ }
- boolean isFinished();
+ @Override
+ public TsBlock receive() {
+ if (throwable != null) {
+ throw new RuntimeException(throwable);
+ }
+ if (closed) {
+ throw new IllegalStateException("Source handle has been closed.");
+ }
+ TsBlock tsBlock = bufferedTsBlocks.poll();
+ if (tsBlock != null) {
+ bufferRetainedSizeInBytes -= getRetainedSizeInBytes(tsBlock);
+ }
+ if (bufferedTsBlocks.isEmpty() && !finished && blocked.isDone()) {
+ blocked = SettableFuture.create();
+ }
+ return tsBlock;
+ }
+
+ private long getRetainedSizeInBytes(TsBlock tsBlock) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isFinished() {
+ return finished;
+ }
- ListenableFuture<Void> isBlocked();
+ public ListenableFuture<Void> isBlocked() {
+ return nonCancellationPropagating(blocked);
+ }
@Override
- void close();
+ public void close() {
+ if (closed) {
+ return;
+ }
+ bufferedTsBlocks.clear();
+ bufferRetainedSizeInBytes = 0;
+ closed = true;
+ if (!blocked.isDone()) {}
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java
b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java
index 98dc0c4..b2086c8 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/schedule/IFragmentInstanceManager.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.db.mpp.schedule;
-import org.apache.iotdb.db.mpp.buffer.IDataBlockManager;
import org.apache.iotdb.db.mpp.schedule.task.FragmentInstanceID;
/** the interface of fragment instance scheduling */
diff --git a/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
b/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
index 9e7e19e..20bfae8 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/ServiceType.java
@@ -67,7 +67,8 @@ public enum ServiceType {
CLUSTER_META_ENGINE("Cluster Meta Engine", "ClusterMetaEngine"),
CLUSTER_DATA_ENGINE("Cluster Data Engine", "ClusterDataEngine"),
REST_SERVICE("REST Service", "REST Service"),
- FRAGMENT_INSTANCE_MANAGER_SERVICE("Fragment instance manager",
"FragmentInstanceManager");
+ FRAGMENT_INSTANCE_MANAGER_SERVICE("Fragment instance manager",
"FragmentInstanceManager"),
+ DATA_BLOCK_MANAGER_SERVICE("Data block manager", "DataBlockManager");
private final String name;
private final String jmxName;
diff --git a/thrift/src/main/thrift/common.thrift
b/thrift/src/main/thrift/mpp.thrift
similarity index 50%
rename from thrift/src/main/thrift/common.thrift
rename to thrift/src/main/thrift/mpp.thrift
index 967deed..7c1f8bc 100644
--- a/thrift/src/main/thrift/common.thrift
+++ b/thrift/src/main/thrift/mpp.thrift
@@ -17,11 +17,39 @@
* under the License.
*/
-namespace java org.apache.iotdb.mpp.common.rpc.thrift
+namespace java org.apache.iotdb.mpp.rpc.thrift
-struct FragmentInstanceID {
- 1: required string queryID
- 2: required string fragmentID
- 3: required string instanceID
+struct FragmentInstanceId {
+ 1: required string queryId
+ 2: required string fragmentId
+ 3: required string instanceId
+}
+
+struct GetDataBlockReqest {
+ 1: required FragmentInstanceId fragnemtInstanceId
+ 2: required i64 blockId
+}
+
+struct GetDataBlockResponse {
+ 1: required list<binary> tsBlocks
+}
+
+struct NewDataBlockEvent {
+ 1: required FragmentInstanceId fragmentInstanceId
+ 2: required string operatorId
+ 3: required i64 blockId
+}
+
+struct EndOfDataBlockEvent {
+ 1: required FragmentInstanceId fragmentInstanceId
+ 2: required string operatorId
+}
+
+service DataBlockService {
+ GetDataBlockResponse getDataBlock(GetDataBlockReqest req);
+
+ void onNewDataBlockEvent(NewDataBlockEvent e);
+
+ void onEndOfDataBlockEvent(EndOfDataBlockEvent e);
}
