IGNITE-3640 IGFS: Local secondary: Investigate whether we need BufferedInputStream for open operation.

modules/core/src/main/java/org/apache/ignite/igfs/secondary/local/LocalIgfsSecondaryFileSystem.java

@@ -255,7 +255,7 @@ private boolean mkdirs0(@Nullable File dir) {
         try {
             FileInputStream in = new FileInputStream(fileForPath(path));
 
-            return new LocalIgfsSecondaryFileSystemPositionedReadable(in, bufSize);
+            return new LocalIgfsSecondaryFileSystemPositionedReadable(in);
         }
         catch (IOException e) {
             throw handleSecondaryFsError(e, "Failed to open file for read: " + path);

modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/secondary/local/LocalIgfsSecondaryFileSystemPositionedReadable.java

@@ -17,49 +17,36 @@
 
 package org.apache.ignite.internal.processors.igfs.secondary.local;
 
-import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystemPositionedReadable;
-
-import java.io.BufferedInputStream;
 import java.io.FileInputStream;
 import java.io.IOException;
+import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystemPositionedReadable;
 
 /**
  * Positioned readable interface for local secondary file system.
  */
-public class LocalIgfsSecondaryFileSystemPositionedReadable extends BufferedInputStream
-    implements IgfsSecondaryFileSystemPositionedReadable {
-    /** Last read position. */
-    private long lastReadPos;
+public class LocalIgfsSecondaryFileSystemPositionedReadable implements IgfsSecondaryFileSystemPositionedReadable {
+    /** File input stream */
+    private final FileInputStream in;
 
     /**
-     * Constructor.
-     *
-     * @param in Input stream.
-     * @param bufSize Buffer size.
+     * @param in FileInputStream
      */
-    public LocalIgfsSecondaryFileSystemPositionedReadable(FileInputStream in, int bufSize) {
-        super(in, bufSize);
+    public LocalIgfsSecondaryFileSystemPositionedReadable(FileInputStream in) {
+        this.in = in;
     }
 
     /** {@inheritDoc} */
     @Override public int read(long readPos, byte[] buf, int off, int len) throws IOException {
         if (in == null)
             throw new IOException("Stream is closed.");
 
-        if (readPos < lastReadPos || readPos + len > lastReadPos + this.buf.length) {
-            ((FileInputStream)in).getChannel().position(readPos);
-
-            pos = 0;
-            count = 0;
-        }
+        in.getChannel().position(readPos);
 
-        int bytesRead = read(buf, off, len);
-
-        if (bytesRead != -1) {
-            // Advance last read position only if we really read some bytes from the stream.
-            lastReadPos = readPos + bytesRead;
-        }
+        return in.read(buf, off, len);
+    }
 
-        return bytesRead;
+    /** {@inheritDoc} */
+    @Override public void close() throws IOException {
+        in.close();
     }
-}
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcRequestHandler.java

@@ -192,9 +192,7 @@ private OdbcResponse executeQuery(long reqId, OdbcQueryExecuteRequest req) {
 
             QueryCursor qryCur = cache.query(qry);
 
-            Iterator iter = qryCur.iterator();
-
-            qryCursors.put(qryId, new IgniteBiTuple<>(qryCur, iter));
+            qryCursors.put(qryId, new IgniteBiTuple<QueryCursor, Iterator>(qryCur, null));
 
             List<?> fieldsMeta = ((QueryCursorImpl) qryCur).fieldsMeta();
 
@@ -220,11 +218,15 @@ private OdbcResponse executeQuery(long reqId, OdbcQueryExecuteRequest req) {
      */
     private OdbcResponse closeQuery(long reqId, OdbcQueryCloseRequest req) {
         try {
-            QueryCursor cur = qryCursors.get(req.queryId()).get1();
+            IgniteBiTuple<QueryCursor, Iterator> tuple = qryCursors.get(req.queryId());
 
-            if (cur == null)
+            if (tuple == null)
                 return new OdbcResponse(OdbcResponse.STATUS_FAILED, "Failed to find query with ID: " + req.queryId());
 
+            QueryCursor cur = tuple.get1();
+
+            assert(cur != null);
+
             cur.close();
 
             qryCursors.remove(req.queryId());
@@ -251,17 +253,27 @@ private OdbcResponse closeQuery(long reqId, OdbcQueryCloseRequest req) {
      */
     private OdbcResponse fetchQuery(long reqId, OdbcQueryFetchRequest req) {
         try {
-            Iterator cur = qryCursors.get(req.queryId()).get2();
+            IgniteBiTuple<QueryCursor, Iterator> tuple = qryCursors.get(req.queryId());
 
-            if (cur == null)
+            if (tuple == null)
                 return new OdbcResponse(OdbcResponse.STATUS_FAILED, "Failed to find query with ID: " + req.queryId());
 
+            Iterator iter = tuple.get2();
+
+            if (iter == null) {
+                QueryCursor cur = tuple.get1();
+
+                iter = cur.iterator();
+
+                tuple.put(cur, iter);
+            }
+
             List<Object> items = new ArrayList<>();
 
-            for (int i = 0; i < req.pageSize() && cur.hasNext(); ++i)
-                items.add(cur.next());
+            for (int i = 0; i < req.pageSize() && iter.hasNext(); ++i)
+                items.add(iter.next());
 
-            OdbcQueryFetchResult res = new OdbcQueryFetchResult(req.queryId(), items, !cur.hasNext());
+            OdbcQueryFetchResult res = new OdbcQueryFetchResult(req.queryId(), items, !iter.hasNext());
 
             return new OdbcResponse(res);
         }

modules/platforms/cpp/binary/src/impl/binary/binary_reader_impl.cpp

@@ -459,8 +459,7 @@ namespace ignite
                     int32_t realLen = stream->ReadInt32();
 
                     if (res && len >= realLen) {
-                        for (int i = 0; i < realLen; i++)
-                            *(res + i) = static_cast<char>(stream->ReadInt8());
+                        stream->ReadInt8Array(reinterpret_cast<int8_t*>(res), realLen);
 
                         if (len > realLen)
                             *(res + realLen) = 0; // Set NULL terminator if possible.

modules/platforms/cpp/odbc/src/connection.cpp

@@ -19,6 +19,8 @@
 
 #include <sstream>
 
+#include <ignite/common/fixed_size_array.h>
+
 #include "ignite/odbc/utility.h"
 #include "ignite/odbc/statement.h"
 #include "ignite/odbc/connection.h"
@@ -178,26 +180,25 @@ namespace ignite
             if (!connected)
                 IGNITE_ERROR_1(IgniteError::IGNITE_ERR_ILLEGAL_STATE, "Connection is not established");
 
-            OdbcProtocolHeader hdr;
+            common::FixedSizeArray<int8_t> msg(len + sizeof(OdbcProtocolHeader));
 
-            hdr.len = static_cast<int32_t>(len);
+            OdbcProtocolHeader *hdr = reinterpret_cast<OdbcProtocolHeader*>(msg.GetData());
 
-            size_t sent = SendAll(reinterpret_cast<int8_t*>(&hdr), sizeof(hdr));
+            hdr->len = static_cast<int32_t>(len);
 
-            if (sent != sizeof(hdr))
-                IGNITE_ERROR_1(IgniteError::IGNITE_ERR_GENERIC, "Can not send message header");
+            memcpy(msg.GetData() + sizeof(OdbcProtocolHeader), data, len);
 
-            sent = SendAll(data, len);
+            size_t sent = SendAll(msg.GetData(), msg.GetSize());
 
-            if (sent != len)
-                IGNITE_ERROR_1(IgniteError::IGNITE_ERR_GENERIC, "Can not send message body");
+            if (sent != len + sizeof(OdbcProtocolHeader))
+                IGNITE_ERROR_1(IgniteError::IGNITE_ERR_GENERIC, "Can not send message");
         }
 
         size_t Connection::SendAll(const int8_t* data, size_t len)
         {
             int sent = 0;
 
-            while (sent != len)
+            while (sent != static_cast<int64_t>(len))
             {
                 int res = socket.Send(data + sent, len - sent);
 
@@ -221,7 +222,7 @@ namespace ignite
 
             OdbcProtocolHeader hdr;
 
-            size_t received = ReceiveAll(reinterpret_cast<int8_t*>(&hdr), sizeof(hdr));
+            int64_t received = ReceiveAll(reinterpret_cast<int8_t*>(&hdr), sizeof(hdr));
 
             if (received != sizeof(hdr))
                 IGNITE_ERROR_1(IgniteError::IGNITE_ERR_GENERIC, "Can not receive message header");