diff --git a/pom.xml b/pom.xml index df1c68f..f6a6fe2 100644 --- a/pom.xml +++ b/pom.xml @@ -27,7 +27,9 @@ UTF-8 yyyy-MM-dd HH:mm:ss - 9.4.54.v20240208 + 9.4.57.v20241219 + 1.8 + 1.8 install @@ -134,6 +136,14 @@ + + org.apache.maven.plugins + maven-compiler-plugin + + true + lines,vars,source + + @@ -149,6 +159,10 @@ org.apache.maven.plugins maven-compiler-plugin + + 9 + 9 + org.apache.maven.plugins @@ -169,7 +183,7 @@ net.sf.jt400 jt400 - 11.2 + 21.0.5 provided @@ -213,6 +227,25 @@ websocket-server ${jetty.version} + + org.mockito + mockito-core + 5.11.0 + test + + + org.junit.jupiter + junit-jupiter-api + 5.10.2 + test + + + org.junit.jupiter + junit-jupiter-engine + 5.10.2 + test + + diff --git a/src/main/java/com/github/ibm/mapepire/BlobRequestData.java b/src/main/java/com/github/ibm/mapepire/BlobRequestData.java new file mode 100644 index 0000000..360eca3 --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/BlobRequestData.java @@ -0,0 +1,26 @@ +package com.github.ibm.mapepire; + +public class BlobRequestData { + + private int replacementIndex; + private int length; + private int offset; + + public BlobRequestData(int replacementIndex, int length, int offset){ + this.replacementIndex = replacementIndex; + this.length = length; + this.offset = offset; + } + + public int getLength() { + return length; + } + + public int getReplacementIndex() { + return replacementIndex; + } + + public int getOffset() { + return offset; + } +} diff --git a/src/main/java/com/github/ibm/mapepire/BlobResponseData.java b/src/main/java/com/github/ibm/mapepire/BlobResponseData.java new file mode 100644 index 0000000..a53f4f6 --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/BlobResponseData.java @@ -0,0 +1,34 @@ +package com.github.ibm.mapepire; + +import java.io.InputStream; +import java.sql.Blob; + +public class BlobResponseData { + final Blob blob; + final String columnName; + final int rowId; + final int length; + + public BlobResponseData(Blob is, String columnName, int rowId, int length){ + this.blob = is; + this.columnName = columnName; + this.rowId = rowId; + this.length = length; + } + + public Blob getBlob() { + return blob; + } + + public String getColumnName(){ + return columnName; + } + + public int getRowId(){ + return rowId; + } + + public int getLength(){ + return length; + } +} diff --git a/src/main/java/com/github/ibm/mapepire/ClientRequest.java b/src/main/java/com/github/ibm/mapepire/ClientRequest.java index 31399cd..9b1b7b9 100644 --- a/src/main/java/com/github/ibm/mapepire/ClientRequest.java +++ b/src/main/java/com/github/ibm/mapepire/ClientRequest.java @@ -16,7 +16,7 @@ public abstract class ClientRequest implements Runnable { private final SystemConnection m_conn; private final String m_id; - private final DataStreamProcessor m_io; + protected final DataStreamProcessor m_io; private final JsonObject m_reqObj; private final Map replyData = new LinkedHashMap(); @@ -28,10 +28,6 @@ protected ClientRequest(final DataStreamProcessor _io, final SystemConnection _c addReplyData("id", m_id); } - public SystemConnection getConnection() { - return m_conn; - } - protected void addReplyData(final String _key, final Object _val) { replyData.put(_key, _val); } @@ -126,7 +122,7 @@ private static String getErrorStringFromException(Throwable _e) { return "Internal Error: " + _e.getClass().getSimpleName(); } - protected void sendreply() throws UnsupportedEncodingException, IOException { + protected void sendreply() throws IOException { final Gson l = new GsonBuilder().serializeNulls().create(); final String json = l.toJson(replyData); replyData.clear(); diff --git a/src/main/java/com/github/ibm/mapepire/DataStreamProcessor.java b/src/main/java/com/github/ibm/mapepire/DataStreamProcessor.java index 462f2be..45604b9 100644 --- a/src/main/java/com/github/ibm/mapepire/DataStreamProcessor.java +++ b/src/main/java/com/github/ibm/mapepire/DataStreamProcessor.java @@ -1,15 +1,17 @@ package com.github.ibm.mapepire; import com.github.ibm.mapepire.requests.*; +import com.github.ibm.mapepire.ws.AsyncSender; import com.github.theprez.jcmdutils.StringUtils; import com.google.gson.JsonElement; import com.google.gson.JsonObject; import com.google.gson.JsonParser; import java.io.*; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.sql.SQLException; -import java.util.HashMap; -import java.util.Map; +import java.util.*; public class DataStreamProcessor implements Runnable { @@ -17,16 +19,19 @@ public class DataStreamProcessor implements Runnable { private final SystemConnection m_conn; private final BufferedReader m_in; - private final PrintStream m_out; + private final PrintStream m_out_text; + private final Map m_queriesMap = new HashMap(); private final Map m_prepStmtMap = new HashMap(); private final boolean m_isTestMode; + private final AsyncSender m_binarySender; - public DataStreamProcessor(final InputStream _in, final PrintStream _out, final SystemConnection _conn, - boolean _isTestMode) + public DataStreamProcessor(final InputStream _in, final PrintStream _outText, final AsyncSender binarySender, final SystemConnection _conn, + boolean _isTestMode) throws UnsupportedEncodingException { m_in = new BufferedReader(new InputStreamReader(_in, "UTF-8")); - m_out = _out; + m_out_text = _outText; + m_binarySender = binarySender; m_conn = _conn; m_isTestMode = _isTestMode; } @@ -66,6 +71,48 @@ public void run() { } } + private int bytesToInt(byte[] bytes, int offset, int length) { + int x = 0; + for (int i = offset; i < offset + length; i++) { + byte b = bytes[i]; + x = x << 8 | b & 0xFF; + } + return x; + } + + + public void run(byte[] payload, int offset, int len) { + int cont_id_length = 2; + int cont_id = bytesToInt(payload, offset, cont_id_length); + + List blobRequestDataArray = new ArrayList<>(); + int curOffset = offset + cont_id_length; + int replacementIndexLength = 1; + int sizeOfInt = 4; + while (curOffset < offset + len) { + int replacementIndex = bytesToInt(payload, curOffset, replacementIndexLength); + curOffset += replacementIndexLength; + + int length = bytesToInt(payload, curOffset, sizeOfInt); + curOffset += sizeOfInt; + + BlobRequestData blobRequestData = new BlobRequestData(replacementIndex, length, curOffset); + blobRequestDataArray.add(blobRequestData); + curOffset += length; + } + + PrepareSql prev = m_prepStmtMap.get(String.valueOf(cont_id)); + if (null == prev) { + dispatch(new BadReq(this, m_conn, null, "invalid correlation ID")); + return; + } + try { + RunBlob runBlob = new RunBlob(payload, blobRequestDataArray, prev); + } catch (Exception e) { + System.out.println("Caught exception " + e); + } + } + public void run(String requestString) { final JsonElement reqElement; final JsonObject reqObj; @@ -179,12 +226,85 @@ public void run(String requestString) { public void sendResponse(final String _response) throws UnsupportedEncodingException, IOException { synchronized (s_replyWriterLock) { - m_out.write((_response + "\n").getBytes("UTF-8")); + m_out_text.write((_response + "\n").getBytes("UTF-8")); Tracer.datastreamOut(_response); - m_out.flush(); + m_out_text.flush(); } } + public void sendResponse(final String id, final BlobResponseData blobResponseData) throws IOException, SQLException { + synchronized (s_replyWriterLock) { + int curOffset = 0; + byte[] buffer = new byte[4 * 1024 * 1024]; + byte[] idBytes = id.getBytes(StandardCharsets.UTF_8); + + curOffset = 0; + String columnName = blobResponseData.getColumnName(); + InputStream is = blobResponseData.getBlob().getBinaryStream(); + int rowId = blobResponseData.getRowId(); + + byte[] columnNameBytes = columnName.getBytes(StandardCharsets.UTF_8); + + if (idBytes.length > 255) { + throw new IllegalArgumentException("ID too long to encode in one byte length"); + } + + // First byte is the length of the ID + buffer[curOffset] = (byte) idBytes.length; + curOffset += 1; + + // Copy ID bytes after the length byte + System.arraycopy(idBytes, 0, buffer, curOffset, idBytes.length); + curOffset += idBytes.length; + + // Copy rowId + ByteBuffer rowIdBuffer = ByteBuffer.allocate(4); + rowIdBuffer.putInt(rowId); // default is big-endian + byte[] rowIdBytes = rowIdBuffer.array(); + for (int j = 0; j < 4; j++) { + buffer[curOffset] = rowIdBytes[j]; + curOffset += 1; + } + + // Copy column length + buffer[curOffset] = (byte) columnName.length(); + curOffset += 1; + + // copy column name + System.arraycopy(columnNameBytes, 0, buffer, curOffset, columnNameBytes.length); + curOffset += columnNameBytes.length; + + // Copy blob length + ByteBuffer blobLength = ByteBuffer.allocate(4); + blobLength.putInt(blobResponseData.getLength()); // default is big-endian + byte[] bytes = blobLength.array(); + for (int j = 0; j < 4; j++) { + buffer[curOffset] = bytes[j]; + curOffset += 1; + } + + int bytesRead = is.read(buffer, curOffset, buffer.length - curOffset); + curOffset += bytesRead; + int totalBytesRead = bytesRead; + if (bytesRead != -1) { + boolean isFinal = totalBytesRead == blobResponseData.getLength(); + sendByteBuffer(buffer, curOffset, isFinal); + } + + while ((bytesRead = is.read(buffer)) != -1) { + totalBytesRead += bytesRead; + boolean isFinal = totalBytesRead == blobResponseData.getLength(); + sendByteBuffer(buffer, bytesRead, isFinal); + } + } + } + + private void sendByteBuffer(byte[] buffer, int bytesRead, boolean isFinal) { + // Wrap only the bytes actually read + ByteBuffer byteBuffer = ByteBuffer.wrap(buffer, 0, bytesRead); + m_binarySender.send(byteBuffer, isFinal); + } + public void end() { try { m_conn.getJdbcConnection().close(); diff --git a/src/main/java/com/github/ibm/mapepire/MapepireServer.java b/src/main/java/com/github/ibm/mapepire/MapepireServer.java index c3971a9..1b20c66 100644 --- a/src/main/java/com/github/ibm/mapepire/MapepireServer.java +++ b/src/main/java/com/github/ibm/mapepire/MapepireServer.java @@ -10,6 +10,7 @@ import javax.net.ssl.TrustManager; import javax.net.ssl.X509TrustManager; +import com.github.ibm.mapepire.ws.DbWebsocketClient; import org.eclipse.jetty.security.ConstraintMapping; import org.eclipse.jetty.security.ConstraintSecurityHandler; import org.eclipse.jetty.server.Server; @@ -39,6 +40,10 @@ public static boolean isSingleMode() { return s_isSingleMode; } + private static int getNumBytesInMb(int num){ + return num * 1024 * 1024; + } + public static void main(final String[] _args) { final LinkedList args = new LinkedList(); @@ -63,9 +68,8 @@ public static void main(final String[] _args) { if (testMode) { System.setIn(new FileInputStream(testFile)); } - final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, conn, testMode); - - io.run(); +// final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, conn, testMode); +// io.run(); } else { s_isSingleMode = false; @@ -170,12 +174,15 @@ public static void main(final String[] _args) { (servletContext, nativeWebSocketConfiguration) -> { nativeWebSocketConfiguration.getPolicy().setMaxTextMessageBufferSize(65535); // Configure max message size - int maxWsMessageSize = 50 * 1024 * 1024; // 50MB + int maxWsMessageSize = getNumBytesInMb(200); + int maxBinaryMessageSize = getNumBytesInMb(200); String maxWsMessageSizeStr = System.getenv("MAX_WS_MESSAGE_SIZE"); if (StringUtils.isNonEmpty(maxWsMessageSizeStr)) { maxWsMessageSize = Integer.parseInt(maxWsMessageSizeStr); } nativeWebSocketConfiguration.getPolicy().setMaxTextMessageSize(maxWsMessageSize); + nativeWebSocketConfiguration.getPolicy().setMaxBinaryMessageSize(maxBinaryMessageSize); + // Add websockets nativeWebSocketConfiguration.addMapping("/db/*", new DbSocketCreator()); diff --git a/src/main/java/com/github/ibm/mapepire/Version.java b/src/main/java/com/github/ibm/mapepire/Version.java index 225954b..c1c6bb0 100644 --- a/src/main/java/com/github/ibm/mapepire/Version.java +++ b/src/main/java/com/github/ibm/mapepire/Version.java @@ -1,5 +1,5 @@ package com.github.ibm.mapepire; public class Version { - static public final String s_compileDateTime = "2024-08-08 00:36:20 (GMT)"; - static public final String s_version = "2.0.0-rc1"; + static public final String s_compileDateTime = "2025-08-25 16:06:39 (GMT)"; + static public final String s_version = "2.3.3"; } \ No newline at end of file diff --git a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java index 911169c..73cc171 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java +++ b/src/main/java/com/github/ibm/mapepire/requests/BlockRetrievableRequest.java @@ -1,29 +1,34 @@ package com.github.ibm.mapepire.requests; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; import java.sql.*; -import java.util.LinkedHashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; +import java.util.*; +import com.github.ibm.mapepire.BlobResponseData; import com.github.ibm.mapepire.ClientRequest; import com.github.ibm.mapepire.DataStreamProcessor; import com.github.ibm.mapepire.SystemConnection; import com.google.gson.JsonObject; +import com.ibm.as400.access.AS400JDBCBlobLocator; import com.ibm.as400.access.AS400JDBCParameterMetaData; +import com.github.ibm.mapepire.BlobResponseData; public abstract class BlockRetrievableRequest extends ClientRequest { protected boolean m_isDone = false; protected ResultSet m_rs = null; protected final boolean m_isTerseData; + private DataStreamProcessor m_io; protected BlockRetrievableRequest(DataStreamProcessor _io, SystemConnection _conn, JsonObject _reqObj) { super(_io, _conn, _reqObj); + m_io = _io; m_isTerseData = getRequestFieldBoolean("terse", false); } - List getNextDataBlock(final int _numRows) throws SQLException { + List getNextDataBlock(final int _numRows) throws SQLException, IOException { if (m_isDone) { return new LinkedList(); } @@ -94,8 +99,8 @@ public Object getData() { } } - protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, final int _numRows, - final boolean _isTerseDataFormat) throws SQLException { + protected DataBlockFetchResult getNextDataBlock(final ResultSet _rs, final int _numRows, + final boolean _isTerseDataFormat) throws SQLException, IOException { final DataBlockFetchResult ret = new DataBlockFetchResult(); if (null == _rs) { @@ -104,6 +109,8 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina if (_rs.isClosed()) { return ret.setDone(true); } + final List blobResponseDataArray = new ArrayList<>(); + int blobsNeeded = 0; for (int i = 0; i < _numRows; ++i) { if (!_rs.next()) { ret.setDone(true); @@ -118,6 +125,8 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina final LinkedHashMap mapRowData = new LinkedHashMap(); final LinkedList terseRowData = new LinkedList(); final int numCols = _rs.getMetaData().getColumnCount(); + int rowId = i; + mapRowData.put("rowId", rowId); for (int col = 1; col <= numCols; ++col) { String column = _rs.getMetaData().getColumnName(col); Object cellData = _rs.getObject(col); @@ -132,7 +141,13 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina } } else if (cellData instanceof Number || cellData instanceof Boolean) { cellDataForResponse = cellData; - } else { + } else if (cellData instanceof Blob){ + blobsNeeded++; + int blobLength = (int) ((Blob) cellData).length(); + BlobResponseData blobResponseData = new BlobResponseData((Blob)cellData, column, rowId, blobLength); + m_io.sendResponse(this.getId(), blobResponseData); + } + else { cellDataForResponse = _rs.getString(col); } if (_isTerseDataFormat) { @@ -142,6 +157,7 @@ protected static DataBlockFetchResult getNextDataBlock(final ResultSet _rs, fina } } ret.add(_isTerseDataFormat ? terseRowData : mapRowData); + addReplyData("blobsNeeded", blobsNeeded); } return ret; } diff --git a/src/main/java/com/github/ibm/mapepire/requests/Exit.java b/src/main/java/com/github/ibm/mapepire/requests/Exit.java index eb77edb..74fe75e 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/Exit.java +++ b/src/main/java/com/github/ibm/mapepire/requests/Exit.java @@ -21,7 +21,7 @@ protected void go() throws Exception { @Override protected void processAfterReplySent() { if (DbSocketCreator.isDaemon()) { - this.getConnection().close(); + this.getSystemConnection().close(); } else { Tracer.info("exiting as requested"); System.exit(0); diff --git a/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java b/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java index 036a74d..4193898 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java +++ b/src/main/java/com/github/ibm/mapepire/requests/PrepareSql.java @@ -19,11 +19,27 @@ public class PrepareSql extends BlockRetrievableRequest { private PreparedStatement m_stmt = null; private final PreparedExecute m_executeTask; + private int blobsNeeded = 0; + private final boolean isImmediateExecute; public PrepareSql(final DataStreamProcessor _io, final SystemConnection m_conn, final JsonObject _reqObj, final boolean _isImmediateExecute) { super(_io, m_conn, _reqObj); - m_executeTask = _isImmediateExecute ? new PreparedExecute(_io, _reqObj, this) : null; + this.isImmediateExecute = _isImmediateExecute; + m_executeTask = new PreparedExecute(_io, _reqObj, this); + this.blobsNeeded = getRequestFieldInt("blobsNeeded", 0); + } + + public int getBlobsNeeded() { + return blobsNeeded; + } + + public PreparedExecute getM_executeTask(){ + return m_executeTask; + } + + public void setBlobsNeeded(int blobsNeeded) { + this.blobsNeeded = blobsNeeded; } @Override @@ -71,13 +87,17 @@ public void go() throws Exception { } addReplyData("metadata", metaData); - if (null != m_executeTask) { - m_executeTask.go(); - this.m_rs = m_executeTask.m_rs; - mergeReplyData(m_executeTask); + if (isImmediateExecute) { + executeTask(); } } + public void executeTask() throws Exception { + m_executeTask.go(); + this.m_rs = m_executeTask.m_rs; + mergeReplyData(m_executeTask); + } + private static String getDb2ParameterName(PreparedStatement _stmt, int _i) throws SQLException { if (_stmt instanceof AS400JDBCPreparedStatement) { return ((AS400JDBCPreparedStatement) _stmt).getDB2ParameterName(_i); @@ -103,7 +123,7 @@ PreparedStatement getStatement() { } @Override public boolean isDone() { - if(null != m_executeTask) { + if (null != m_executeTask) { return m_isDone || m_executeTask.isDone(); } // This means that the request was _only_ for a prepare, so yes we're done if we haven't fetched any data diff --git a/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java index 0f7a6a4..6057e20 100644 --- a/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java +++ b/src/main/java/com/github/ibm/mapepire/requests/PreparedExecute.java @@ -1,11 +1,20 @@ package com.github.ibm.mapepire.requests; +import java.io.ByteArrayInputStream; +import java.io.StringReader; +import java.math.BigDecimal; import java.sql.CallableStatement; import java.sql.ParameterMetaData; import java.sql.PreparedStatement; import java.sql.SQLException; +import java.sql.Date; +import java.sql.Time; +import java.sql.Timestamp; + + import java.sql.Types; import java.util.Arrays; +import java.util.Base64; import java.util.LinkedList; import com.github.ibm.mapepire.DataStreamProcessor; @@ -25,6 +34,13 @@ public PreparedExecute(final DataStreamProcessor _io, final JsonObject _reqObj, @Override protected void go() throws Exception { JsonArray parms = super.getRequestField("parameters").getAsJsonArray(); + JsonElement columnTypesElement = super.getRequestField("columnTypes"); + JsonArray columnTypes; + if (columnTypesElement != null){ + columnTypes = columnTypesElement.getAsJsonArray(); + } else { + columnTypes = null; + } boolean isBatch = !parms.isEmpty() && parms.get(0).isJsonArray(); boolean hasResultSet = false; long batchUpdateCount = 0; @@ -34,7 +50,7 @@ protected void go() throws Exception { JsonArray arr = parms.getAsJsonArray(); int batch_ops_added = 0; for (int i = 0; i < arr.size(); i++) { - addJsonArrayParameters(stmt, arr.get(i).getAsJsonArray()); + addJsonArrayParameters(stmt, arr.get(i).getAsJsonArray(), columnTypes); m_prev.getStatement().addBatch(); } batch_ops_added += arr.size(); @@ -43,7 +59,7 @@ protected void go() throws Exception { batchUpdateCount = Arrays.stream(updateCount).sum(); } else{ if (parms != null) { - addJsonArrayParameters(stmt, parms.getAsJsonArray()); + addJsonArrayParameters(stmt, parms.getAsJsonArray(), columnTypes); } hasResultSet = stmt.execute(); } @@ -66,9 +82,87 @@ protected void go() throws Exception { } } - private void addJsonArrayParameters(PreparedStatement stmt, JsonArray arr) throws SQLException { - for (int i = 1; i <= arr.size(); ++i) { - JsonElement element = arr.get(-1 + i); + private void addParameter(int i, JsonElement parameter, ColumnType columnType, PreparedStatement stmt) throws SQLException { + String stringValue = parameter.getAsString(); + + switch (columnType) { + case BLOB: { + byte[] decodedBytes = Base64.getDecoder().decode(stringValue); + ByteArrayInputStream blob = new ByteArrayInputStream(decodedBytes); + stmt.setBlob(i, blob); + break; + } + case CLOB: + stmt.setClob(i, new StringReader(stringValue)); + break; + case INTEGER: + stmt.setInt(i, Integer.parseInt(stringValue)); + break; + case BIGINT: + stmt.setLong(i, Long.parseLong(stringValue)); + break; + case SMALLINT: + stmt.setShort(i, Short.parseShort(stringValue)); + break; + case DOUBLE: + stmt.setDouble(i, Double.parseDouble(stringValue)); + break; + case FLOAT: + stmt.setFloat(i, Float.parseFloat(stringValue)); + break; + case DECIMAL: + stmt.setBigDecimal(i, new BigDecimal(stringValue)); + break; + case DATE: + stmt.setDate(i, Date.valueOf(stringValue)); // expects ISO format: yyyy-[m]m-[d]d + break; + case TIME: + stmt.setTime(i, Time.valueOf(stringValue)); // expects format: hh:mm:ss + break; + case TIMESTAMP: + stmt.setTimestamp(i, Timestamp.valueOf(stringValue)); // expects format: yyyy-[m]m-[d]d hh:mm:ss[.f...] + break; + case BOOLEAN: + stmt.setBoolean(i, Boolean.parseBoolean(stringValue)); + break; + default: + stmt.setString(i, stringValue); // Fallback + } + } + + + public enum ColumnType { + CHAR, + VARCHAR, + INTEGER, + FLOAT, + DECIMAL, + DATE, + TIME, + TIMESTAMP, + CLOB, + BLOB, + BOOLEAN, + SMALLINT, + BIGINT, + DOUBLE + } + + private ColumnType getColumnType(int i, JsonArray columnTypes){ + ColumnType columnType; + try { + columnType = ColumnType.valueOf(columnTypes.get(i).getAsString()); + } catch (IllegalArgumentException e){ + columnType = ColumnType.VARCHAR; + } + return columnType; + } + + private void addJsonArrayParameters(PreparedStatement stmt, JsonArray parameters, JsonArray columnTypes) throws SQLException { + for (int i = 1; i <= parameters.size(); ++i) { + JsonElement element = parameters.get(-1 + i); + ColumnType columnType = columnTypes == null ? ColumnType.VARCHAR : getColumnType(-1 + i, columnTypes); + if (element.isJsonNull()) { stmt.setNull(i, Types.NULL); } else { @@ -78,9 +172,9 @@ private void addJsonArrayParameters(PreparedStatement stmt, JsonArray arr) throw } else if (stmt instanceof CallableStatement && ParameterMetaData.parameterModeInOut == stmt.getParameterMetaData().getParameterMode(i)) { ((CallableStatement) stmt).registerOutParameter(i, stmt.getParameterMetaData().getParameterType(i)); - stmt.setString(i, element.getAsString()); + addParameter(i, element, columnType, stmt); } else { - stmt.setString(i, element.getAsString()); + addParameter(i, element, columnType, stmt); } } } diff --git a/src/main/java/com/github/ibm/mapepire/requests/RunBlob.java b/src/main/java/com/github/ibm/mapepire/requests/RunBlob.java new file mode 100644 index 0000000..bcf98d5 --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/requests/RunBlob.java @@ -0,0 +1,36 @@ +package com.github.ibm.mapepire.requests; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.sql.*; +import java.util.Arrays; +import java.util.LinkedList; +import java.util.List; + +import com.github.ibm.mapepire.BlobRequestData; +import com.github.ibm.mapepire.DataStreamProcessor; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; + +import javax.sql.rowset.serial.SerialBlob; + +public class RunBlob{ + + private final PrepareSql m_prev; + + public RunBlob(final byte[] binary, final List blobRequestDataList, final PrepareSql _prev) throws Exception { + m_prev = _prev; + PreparedStatement stmt = m_prev.getStatement(); + for (BlobRequestData blobRequestData: blobRequestDataList){ + int offset = blobRequestData.getOffset(); + int length = blobRequestData.getLength(); + int replacementIndex = blobRequestData.getReplacementIndex(); + InputStream is = new ByteArrayInputStream(binary, offset, length); + stmt.setBinaryStream(replacementIndex, is, length); + + } + _prev.executeTask(); + } +} diff --git a/src/main/java/com/github/ibm/mapepire/ws/AsyncSender.java b/src/main/java/com/github/ibm/mapepire/ws/AsyncSender.java new file mode 100644 index 0000000..3805326 --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/ws/AsyncSender.java @@ -0,0 +1,41 @@ +package com.github.ibm.mapepire.ws; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import org.eclipse.jetty.websocket.api.Session; +import org.eclipse.jetty.websocket.api.RemoteEndpoint; +import org.eclipse.jetty.websocket.api.WebSocketAdapter; + +public class AsyncSender { + private final ExecutorService executor = Executors.newFixedThreadPool(1); + private final WebSocketAdapter session; + + public AsyncSender(WebSocketAdapter session) { + this.session = session; + } + + public void send(ByteBuffer buffer, boolean isLast) { + try { + + executor.submit(() -> { + RemoteEndpoint remote = session.getRemote(); + try { + remote.sendPartialBytes(buffer, isLast); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + ).get();} + catch (Exception e) { + e.printStackTrace(); // or better logging + } + } + + public void shutdown() { + executor.shutdown(); + } +} diff --git a/src/main/java/com/github/ibm/mapepire/ws/BinarySender.java b/src/main/java/com/github/ibm/mapepire/ws/BinarySender.java new file mode 100644 index 0000000..503722a --- /dev/null +++ b/src/main/java/com/github/ibm/mapepire/ws/BinarySender.java @@ -0,0 +1,9 @@ +package com.github.ibm.mapepire.ws; + +import java.io.IOException; +import java.nio.ByteBuffer; + +@FunctionalInterface +public interface BinarySender { + void send(ByteBuffer buffer, boolean isLast) throws IOException; +} \ No newline at end of file diff --git a/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java b/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java index 6dc0986..9edd6e8 100644 --- a/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java +++ b/src/main/java/com/github/ibm/mapepire/ws/DbWebsocketClient.java @@ -7,7 +7,6 @@ import org.eclipse.jetty.websocket.api.WebSocketException; import java.io.*; -import java.nio.ByteBuffer; import java.util.concurrent.CountDownLatch; public class DbWebsocketClient extends WebSocketAdapter { @@ -17,7 +16,7 @@ public class DbWebsocketClient extends WebSocketAdapter { DbWebsocketClient(String clientHost, String clientAddress, String host, String user, String pass) throws IOException { super(); SystemConnection conn = new SystemConnection(clientHost, clientAddress,host, user, pass); - io = getDataStream(this, conn); + io = getDataStreamProcessor(this, conn); } @Override @@ -33,6 +32,11 @@ public void onWebSocketText(String message) { io.run(message); } + @Override + public void onWebSocketBinary(byte[] payload, int offset, int len) { + io.run(payload, offset, len); + } + @Override public void onWebSocketClose(int statusCode, String reason) { io.end(); @@ -51,10 +55,12 @@ public void awaitClosure() throws InterruptedException { closureLatch.await(); } - private static DataStreamProcessor getDataStream(DbWebsocketClient endpoint, SystemConnection conn) throws UnsupportedEncodingException { + private DataStreamProcessor getDataStreamProcessor(DbWebsocketClient endpoint, SystemConnection conn) throws UnsupportedEncodingException { + AsyncSender asyncSender = new AsyncSender(this); + InputStream in = new ByteArrayInputStream(new byte[0]); - OutputStream outStream = new OutputStream() { + OutputStream outStreamText = new OutputStream() { private final ByteArrayOutputStream payload = new ByteArrayOutputStream(); @Override @@ -82,8 +88,8 @@ public synchronized void flush() throws IOException { } }; - PrintStream out = new PrintStream(outStream); + PrintStream out = new PrintStream(outStreamText); - return new DataStreamProcessor(in, out, conn, false); + return new DataStreamProcessor(in, out, asyncSender, conn, false); } } diff --git a/src/test/java/BlobTest.java b/src/test/java/BlobTest.java new file mode 100644 index 0000000..399482a --- /dev/null +++ b/src/test/java/BlobTest.java @@ -0,0 +1,547 @@ +import com.github.ibm.mapepire.DataStreamProcessor; +import com.github.ibm.mapepire.requests.BlockRetrievableRequest; +import com.github.ibm.mapepire.ws.AsyncSender; +//import com.github.ibm.mapepire.ws.asyncSender; +import com.google.gson.JsonObject; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import static org.mockito.Mockito.*; +import static org.junit.jupiter.api.Assertions.*; +import javax.sql.rowset.serial.SerialBlob; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.sql.Blob; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.*; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.mock; + +public class BlobTest { + + @Test + public void testMockResultSetSingleRow() throws SQLException { + // Arrange + Map row = new HashMap<>(); + List> rowsByIndex; + + row.put("id", 1); + row.put("name", "Alice"); + + List> rows = Collections.singletonList(row); + ResultSet rs = new MockResultSet(rows, null); + + // Act & Assert + assertTrue(rs.next(), "Expected at least one row"); + assertEquals(1, rs.getInt("id")); + assertEquals("Alice", rs.getString("name")); + + assertFalse(rs.next(), "No more rows expected"); + } + + @Test + public void testMockResultSetMultipleRows() throws SQLException { + // Arrange + Map row1 = new HashMap<>(); + row1.put("id", 1); + row1.put("name", "Alice"); + + Map row2 = new HashMap<>(); + row2.put("id", 2); + row2.put("name", "Bob"); + + List> rows = Arrays.asList(row1, row2); + ResultSet rs = new MockResultSet(rows, null); + + // First row + assertTrue(rs.next()); + assertEquals(1, rs.getInt("id")); + assertEquals("Alice", rs.getString("name")); + + // Second row + assertTrue(rs.next()); + assertEquals(2, rs.getInt("id")); + assertEquals("Bob", rs.getString("name")); + + // No more + assertFalse(rs.next()); + } + + @Test + void testBlobColumnInResultSet() throws Exception { + // Create some fake binary data + byte[] data = "hello world".getBytes(); + + // Wrap it in a Blob + Blob blob = new SerialBlob(data); + + List> rowsByIndex = Arrays.asList(Arrays.asList(blob)); + + + // Use MockResultSet to simulate a DB result + MockResultSet rs = new MockResultSet(rowsByIndex); + + // Act: move to the first row + assertTrue(rs.next()); + + // Retrieve the blob + Blob retrievedBlob = rs.getBlob(1); + assertNotNull(retrievedBlob); + + // Read the bytes back + byte[] retrievedData = retrievedBlob.getBytes(1, (int) retrievedBlob.length()); + + // Verify the round-trip matches + assertArrayEquals(data, retrievedData); + + // No more rows + assertFalse(rs.next()); + } + +// @Test +// public void testGetBinaryStreamByIndex() throws SQLException, IOException { +// // Prepare blob data +// byte[] blobData = {1, 2, 3, 4, 5}; +// +// // Use rowsByIndex to support getBinaryStream(int columnIndex) +// List> rowsByIndex = new ArrayList<>(); +// rowsByIndex.add(Arrays.asList("Alice", new ByteArrayInputStream(blobData))); +// +// MockResultSet rs = new MockResultSet(rowsByIndex); +// +// assertTrue(rs.next(), "Should have first row"); +// +// InputStream is = rs.getBinaryStream(2); // second column +// assertNotNull(is, "BinaryStream should not be null"); +// +// byte[] readData = is.readAllBytes(); +// assertArrayEquals(blobData, readData, "Binary data should match"); +// } + + @Test + public void testSendingSingleRowWithBlob() throws SQLException, IOException { + JsonObject obj = new JsonObject(); + obj.addProperty("id", "12345"); + AsyncSender asyncSender = mock(AsyncSender.class); + final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, asyncSender, null, true); + + + BlockRetrievableRequestImpl block = new BlockRetrievableRequestImpl(io, null, obj); + byte[] blobData = {1, 2, 3, 4, 5}; + byte[] expectedResult = {5, 49, 50, 51, 52, 53, 0,0,0,0, 4, 98, 108, 111, 98, 0,0,0,5, 1, 2, 3, 4, 5}; // queryIdLength | queryId | rowId | colNameLength | colName| bloblLength| blob + Blob blob = new SerialBlob(blobData); + + // Use rowsByIndex to support getBinaryStream(int columnIndex) + List> rowsByIndex = new ArrayList<>(); + rowsByIndex.add(Arrays.asList(blob)); + + Map row1 = new HashMap<>(); + row1.put("blob", blob); + List> rows = Arrays.asList(row1); + MockResultSet rs = new MockResultSet(rows, rowsByIndex); + block.getNextDataBlockUsage(rs, 1, false); + + verify(asyncSender, times(1)).send(any(ByteBuffer.class), anyBoolean()); + + // Assert - capture arguments + ArgumentCaptor bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); + ArgumentCaptor booleanCaptor = ArgumentCaptor.forClass(Boolean.class); + + verify(asyncSender, times(1)).send(bufferCaptor.capture(), booleanCaptor.capture()); + + // Verify first call + ByteBuffer firstBuf = bufferCaptor.getAllValues().get(0); + Boolean firstFlag = booleanCaptor.getAllValues().get(0); + byte[] actual = Arrays.copyOfRange(firstBuf.array(), firstBuf.position(), firstBuf.limit()); + assertArrayEquals(expectedResult, actual); + assertTrue(firstFlag); + } + + @Test + public void testSendingMultipleRowsWithBlobs() throws SQLException, IOException { + JsonObject obj = new JsonObject(); + obj.addProperty("id", "12345"); + AsyncSender asyncSender = mock(AsyncSender.class); + final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, asyncSender, null, true); + + + BlockRetrievableRequestImpl block = new BlockRetrievableRequestImpl(io, null, obj); + byte[] blob1Data = {1, 2, 3, 4, 5}; + byte[] blob2Data = {6, 7, 8}; + + byte[] expectedResult1 = {5, 49, 50, 51, 52, 53, 0,0,0,0, 4, 98, 108, 111, 98, 0,0,0,5, 1, 2, 3, 4, 5}; // queryIdLength | queryId | rowId | colNameLength | colName| bloblLength| blob + byte[] expectedResult2 = {5, 49, 50, 51, 52, 53, 0,0,0,1, 4, 98, 108, 111, 98, 0,0,0,3, 6, 7, 8}; // rowId | colNameLength | colName| bloblLength| blob + + Blob blob1 = new SerialBlob(blob1Data); + Blob blob2 = new SerialBlob(blob2Data); + + // Use rowsByIndex to support getBinaryStream(int columnIndex) + List> rowsByIndex = new ArrayList<>(); + rowsByIndex.add(Arrays.asList(blob1)); + rowsByIndex.add(Arrays.asList(blob2)); + + Map row1 = new HashMap<>(); + Map row2 = new HashMap<>(); + + row1.put("blob", blob1); + row2.put("blob", blob2); + + List> rows = Arrays.asList(row1, row2); + MockResultSet rs = new MockResultSet(rows, rowsByIndex); + block.getNextDataBlockUsage(rs, 2, false); + + verify(asyncSender, times(2)).send(any(ByteBuffer.class), anyBoolean()); + + // Assert - capture arguments + ArgumentCaptor bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); + ArgumentCaptor booleanCaptor = ArgumentCaptor.forClass(Boolean.class); + + verify(asyncSender, times(2)).send(bufferCaptor.capture(), booleanCaptor.capture()); + + // Verify first call + ByteBuffer firstBuf = bufferCaptor.getAllValues().get(0); + Boolean firstFlag = booleanCaptor.getAllValues().get(0); + byte[] actual = Arrays.copyOfRange(firstBuf.array(), firstBuf.position(), firstBuf.limit()); + assertArrayEquals(expectedResult1, actual); + assertTrue(firstFlag); + + // Verify second call + ByteBuffer secondBuf = bufferCaptor.getAllValues().get(1); + Boolean secondFlag = booleanCaptor.getAllValues().get(1); + byte[] actualSecond = Arrays.copyOfRange(secondBuf.array(), secondBuf.position(), secondBuf.limit()); + assertArrayEquals(expectedResult2, actualSecond); + assertTrue(secondFlag); + } + + + @Test + public void testSingleRowMultipleBlobs() throws SQLException, IOException { + JsonObject obj = new JsonObject(); + obj.addProperty("id", "123"); + AsyncSender asyncSender = mock(AsyncSender.class); + final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, asyncSender, null, true); + + + BlockRetrievableRequestImpl block = new BlockRetrievableRequestImpl(io, null, obj); + byte[] blobData1 = {10, 11}; + byte[] blobData2 = {20, 21, 22}; + Blob blob1 = new SerialBlob(blobData1); + Blob blob2 = new SerialBlob(blobData2); + + + + byte[] expected1 = { + 3, '1','2','3', // queryId + 0,0,0,0, // rowId + 1, 'a', // colName "a" + 0,0,0,2, 10, 11, // blob + }; + + byte[] expected2 = { + 3, '1','2','3', // queryId + 0,0,0,0, // rowId + 2, 'b','b', // colName "bb" + 0,0,0,3, 20, 21, 22 // blob + }; + // Use rowsByIndex to support getBinaryStream(int columnIndex) + List> rowsByIndex = new ArrayList<>(); + rowsByIndex.add(Arrays.asList(blob1, blob2)); + + Map row1 = new HashMap<>(); + + row1.put("a", blob1); + row1.put("bb", blob2); + + List> rows = Arrays.asList(row1); + MockResultSet rs = new MockResultSet(rows, rowsByIndex); + block.getNextDataBlockUsage(rs, 1, false); + + verify(asyncSender, times(2)).send(any(ByteBuffer.class), anyBoolean()); + + // Assert - capture arguments + ArgumentCaptor bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); + ArgumentCaptor booleanCaptor = ArgumentCaptor.forClass(Boolean.class); + + verify(asyncSender, times(2)).send(bufferCaptor.capture(), booleanCaptor.capture()); + + // Verify first call + ByteBuffer firstBuf = bufferCaptor.getAllValues().get(0); + Boolean firstFlag = booleanCaptor.getAllValues().get(0); + byte[] actual = Arrays.copyOfRange(firstBuf.array(), firstBuf.position(), firstBuf.limit()); + assertArrayEquals(expected1, actual); + assertTrue(firstFlag); + + // Verify second call + ByteBuffer secondBuf = bufferCaptor.getAllValues().get(1); + Boolean secondFlag = booleanCaptor.getAllValues().get(1); + byte[] actualSecond = Arrays.copyOfRange(secondBuf.array(), secondBuf.position(), secondBuf.limit()); + assertArrayEquals(expected2, actualSecond); + assertTrue(secondFlag); + } + + @Test + public void testMultipleRowsMultipleBlobsPerRow() throws SQLException, IOException { + JsonObject obj = new JsonObject(); + obj.addProperty("id", "query7"); + AsyncSender asyncSender = mock(AsyncSender.class); + final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, asyncSender, null, true); + + + BlockRetrievableRequestImpl block = new BlockRetrievableRequestImpl(io, null, obj); + byte[] blob1Data = {1, 2, 3, 4, 5}; + byte[] blob2Data = {6, 7, 8}; + + byte[] expected1 = { + 6, 'q', 'u', 'e', 'r', 'y', '7', // queryId + 0,0,0,0, // rowId + 11, 'b', 'l','o','b','C','o','l','u','m','n','1', // colName "a" + 0,0,0,5, 1, 2, 3, 4 ,5 // blob + }; + + byte[] expected2 = { + 6, 'q', 'u', 'e', 'r', 'y', '7', // queryId + 0,0,0,0, // rowId + 11, 'b', 'l','o','b','C','o','l','u','m','n','2', + 0,0,0,3, 6, 7, 8 // blob + }; + + byte[] expected3 = { + 6, 'q', 'u', 'e', 'r', 'y', '7', // queryId + 0,0,0,1, // rowId + 11, 'b', 'l','o','b','C','o','l','u','m','n','1', + 0,0,0,5, 1, 2, 3, 4 ,5 // blob + }; + + byte[] expected4 = { + 6, 'q', 'u', 'e', 'r', 'y', '7', // queryId + 0,0,0,1, // rowId + 11, 'b', 'l','o','b','C','o','l','u','m','n','2', + 0,0,0,3, 6, 7, 8 // blob + }; + + + Blob blob1 = new SerialBlob(blob1Data); + Blob blob2 = new SerialBlob(blob2Data); + + // Use rowsByIndex to support getBinaryStream(int columnIndex) + List> rowsByIndex = new ArrayList<>(); + rowsByIndex.add(Arrays.asList(blob1, blob2)); + rowsByIndex.add(Arrays.asList(blob1, blob2)); + + Map row1 = new HashMap<>(); + Map row2 = new HashMap<>(); + + row1.put("blobColumn1", blob1); + row1.put("blobColumn2", blob2); + + row2.put("blobColumn1", blob1); + row2.put("blobColumn2", blob2); + + + List> rows = Arrays.asList(row1, row2); + MockResultSet rs = new MockResultSet(rows, rowsByIndex); + block.getNextDataBlockUsage(rs, 2, false); + + verify(asyncSender, times(4)).send(any(ByteBuffer.class), anyBoolean()); + + // Assert - capture arguments + ArgumentCaptor bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); + ArgumentCaptor booleanCaptor = ArgumentCaptor.forClass(Boolean.class); + + verify(asyncSender, times(4)).send(bufferCaptor.capture(), booleanCaptor.capture()); + + // Verify first call + ByteBuffer firstBuf = bufferCaptor.getAllValues().get(0); + Boolean firstFlag = booleanCaptor.getAllValues().get(0); + byte[] actual = Arrays.copyOfRange(firstBuf.array(), firstBuf.position(), firstBuf.limit()); + assertArrayEquals(expected1, actual); + assertTrue(firstFlag); + + // Verify second call + ByteBuffer secondBuf = bufferCaptor.getAllValues().get(1); + Boolean secondFlag = booleanCaptor.getAllValues().get(1); + byte[] actualSecond = Arrays.copyOfRange(secondBuf.array(), secondBuf.position(), secondBuf.limit()); + assertArrayEquals(expected2, actualSecond); + assertTrue(secondFlag); + + // Verify third call + ByteBuffer thirdBuf = bufferCaptor.getAllValues().get(2); + Boolean thirdFlag = booleanCaptor.getAllValues().get(2); + byte[] actualThird = Arrays.copyOfRange(thirdBuf.array(), thirdBuf.position(), thirdBuf.limit()); + assertArrayEquals(expected3, actualThird); + assertTrue(thirdFlag); + + // Verify fourth call + ByteBuffer fourthBuf = bufferCaptor.getAllValues().get(3); + Boolean fourthFlag = booleanCaptor.getAllValues().get(3); + byte[] actualFourth = Arrays.copyOfRange(fourthBuf.array(), fourthBuf.position(), fourthBuf.limit()); + assertArrayEquals(expected4, actualFourth); + assertTrue(fourthFlag); + } + + + @Test + public void testSendingSingleRowWithEmptyBlob() throws SQLException, IOException { + JsonObject obj = new JsonObject(); + obj.addProperty("id", "12345"); + AsyncSender asyncSender = mock(AsyncSender.class); + final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, asyncSender, null, true); + + + BlockRetrievableRequestImpl block = new BlockRetrievableRequestImpl(io, null, obj); + byte[] blobData = {}; + Blob blob = new SerialBlob(blobData); + + // Use rowsByIndex to support getBinaryStream(int columnIndex) + List> rowsByIndex = new ArrayList<>(); + rowsByIndex.add(Arrays.asList(blob)); + + Map row1 = new HashMap<>(); + row1.put("blob", blob); + List> rows = Arrays.asList(row1); + MockResultSet rs = new MockResultSet(rows, rowsByIndex); + block.getNextDataBlockUsage(rs, 1, false); + + verify(asyncSender, times(0)).send(any(ByteBuffer.class), anyBoolean()); + + // Assert - capture arguments + ArgumentCaptor bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); + ArgumentCaptor booleanCaptor = ArgumentCaptor.forClass(Boolean.class); + + verify(asyncSender, times(0)).send(bufferCaptor.capture(), booleanCaptor.capture()); + + } + + @Test + public void testSendingSingleRowWithBlobandNonblobColumns() throws SQLException, IOException { + JsonObject obj = new JsonObject(); + obj.addProperty("id", "12345"); + AsyncSender asyncSender = mock(AsyncSender.class); + final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, asyncSender, null, true); + + + BlockRetrievableRequestImpl block = new BlockRetrievableRequestImpl(io, null, obj); + byte[] blobData = {1, 2, 3, 4, 5}; + String strCol = "Some string column data"; + byte[] expectedResult = {5, 49, 50, 51, 52, 53, 0,0,0,0, 4, 98, 108, 111, 98, 0,0,0,5, 1, 2, 3, 4, 5}; // queryIdLength | queryId | rowId | colNameLength | colName| bloblLength| blob + Blob blob = new SerialBlob(blobData); + + // Use rowsByIndex to support getBinaryStream(int columnIndex) + List> rowsByIndex = new ArrayList<>(); + + rowsByIndex.add(Arrays.asList(blob, strCol)); + + Map row1 = new HashMap<>(); + row1.put("blob", blob); + row1.put("strCol", strCol); + + List> rows = Arrays.asList(row1); + MockResultSet rs = new MockResultSet(rows, rowsByIndex); + block.getNextDataBlockUsage(rs, 1, false); + + verify(asyncSender, times(1)).send(any(ByteBuffer.class), anyBoolean()); + + // Assert - capture arguments + ArgumentCaptor bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); + ArgumentCaptor booleanCaptor = ArgumentCaptor.forClass(Boolean.class); + + verify(asyncSender, times(1)).send(bufferCaptor.capture(), booleanCaptor.capture()); + + // Verify first call + ByteBuffer firstBuf = bufferCaptor.getAllValues().get(0); + Boolean firstFlag = booleanCaptor.getAllValues().get(0); + byte[] actual = Arrays.copyOfRange(firstBuf.array(), firstBuf.position(), firstBuf.limit()); + assertArrayEquals(expectedResult, actual); + assertTrue(firstFlag); + } + + + @Test + public void testSendingSingleRowWithLongBlob() throws SQLException, IOException { + JsonObject obj = new JsonObject(); + obj.addProperty("id", "12345"); + AsyncSender asyncSender = mock(AsyncSender.class); + final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, asyncSender, null, true); + + + BlockRetrievableRequestImpl block = new BlockRetrievableRequestImpl(io, null, obj); + int blobSize = 5 * 1024 * 1024; // 5 MB + byte[] blobData = new byte[blobSize]; + for (int i = 0; i < blobSize; i++) { + blobData[i] = (byte) (i % 256); // repeat 0-255 + } + +// byte[] expected1 = { +// 6, 'q', 'u', 'e', 'r', 'y', '7', // queryId +// 0, // rowId +// 11, 'b', 'l','o','b','C','o','l','u','m','n','1', // colName "a" +// 5, 1, 2, 3, 4 ,5 // blob +// }; + byte[] expectedResult = { + 5, 49, 50, 51, 52, 53, // querylength, id + 0,0,0,0, // row id + 4, 98, 108, 111, 98, // col name length, colname + 0, 80, 0, 0, 0, 1, 2, 3, 4, 5}; // blob length, blob + + byte[] expectedResult2 = { + -19, -18, -17, -16, -15}; // blob length, blob + Blob blob = new SerialBlob(blobData); + + // Use rowsByIndex to support getBinaryStream(int columnIndex) + List> rowsByIndex = new ArrayList<>(); + rowsByIndex.add(Arrays.asList(blob)); + + Map row1 = new HashMap<>(); + row1.put("blob", blob); + List> rows = Arrays.asList(row1); + MockResultSet rs = new MockResultSet(rows, rowsByIndex); +// +// verify(asyncSender, times(641)).send(any(ByteBuffer.class), anyBoolean()); +// +// // Assert - capture arguments +// ArgumentCaptor bufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); +// ArgumentCaptor booleanCaptor = ArgumentCaptor.forClass(Boolean.class); +// +// verify(asyncSender, times(641)).send(bufferCaptor.capture(), booleanCaptor.capture()); +// +// // Verify first call +// ByteBuffer firstBuf = bufferCaptor.getAllValues().get(0); +// Boolean firstFlag = booleanCaptor.getAllValues().get(0); +// byte[] actual = Arrays.copyOfRange(firstBuf.array(), firstBuf.position(), expectedResult.length); +// assertArrayEquals(expectedResult, actual); +// assertTrue(firstFlag); + + List capturedBuffers = new ArrayList<>(); + List capturedFlags = new ArrayList<>(); + + doAnswer(invocation -> { + ByteBuffer buf = invocation.getArgument(0); + boolean isFinal = invocation.getArgument(1); + + // Copy the bytes immediately, so later modifications don't affect this snapshot + byte[] snapshot = Arrays.copyOfRange(buf.array(), buf.position(), buf.limit()); + capturedBuffers.add(snapshot); + capturedFlags.add(isFinal); + + return null; + }).when(asyncSender).send(any(ByteBuffer.class), anyBoolean()); + block.getNextDataBlockUsage(rs, 1, false); + + byte[] actualFirst21 = Arrays.copyOfRange(capturedBuffers.get(0), 0, 25); + byte[] actualSecondBuf = Arrays.copyOfRange(capturedBuffers.get(1), 0, 5); + + + assertArrayEquals(expectedResult, actualFirst21); + assertArrayEquals(expectedResult2, actualSecondBuf); + + assertFalse(capturedFlags.get(0)); + + + + } + + +} diff --git a/src/test/java/BlockRetrievableRequestImpl.java b/src/test/java/BlockRetrievableRequestImpl.java new file mode 100644 index 0000000..6eafa68 --- /dev/null +++ b/src/test/java/BlockRetrievableRequestImpl.java @@ -0,0 +1,22 @@ +import com.github.ibm.mapepire.DataStreamProcessor; +import com.github.ibm.mapepire.SystemConnection; +import com.github.ibm.mapepire.requests.BlockRetrievableRequest; +import com.google.gson.JsonObject; + +import java.io.IOException; +import java.sql.ResultSet; +import java.sql.SQLException; + +public class BlockRetrievableRequestImpl extends BlockRetrievableRequest { + protected BlockRetrievableRequestImpl(DataStreamProcessor _io, SystemConnection _conn, JsonObject _reqObj) { + super(_io, _conn, _reqObj); + } + + public DataBlockFetchResult getNextDataBlockUsage(final ResultSet _rs, final int _numRows, + final boolean _isTerseDataFormat) throws SQLException, IOException { + return getNextDataBlock(_rs, _numRows, _isTerseDataFormat); + } + @Override + protected void go() throws Exception { + } +} diff --git a/src/test/java/MockResultSet.java b/src/test/java/MockResultSet.java new file mode 100644 index 0000000..064e571 --- /dev/null +++ b/src/test/java/MockResultSet.java @@ -0,0 +1,1033 @@ +import java.io.InputStream; +import java.io.Reader; +import java.math.BigDecimal; +import java.net.URL; +import java.sql.*; +import java.sql.Date; +import java.util.*; + +public class MockResultSet implements ResultSet { + private List> rows = null; + private List> rowsByIndex = null; + + private int cursor = -1; + + public MockResultSet(List> rows, List> rowsByIndex) { + this.rows = rows; + this.rowsByIndex = rowsByIndex; + } + + + public MockResultSet( List> rowsByIndex) { + this.rowsByIndex = rowsByIndex; + } + + @Override + public ResultSetMetaData getMetaData() throws SQLException { + return new ResultSetMetaData() { + private final List columns = new ArrayList<>(rows.get(0).keySet()); + { // instance initializer + Collections.sort(columns); + } + + @Override + public int getColumnCount() throws SQLException { + return columns.size(); + } + + @Override + public String getColumnName(int column) throws SQLException { + // JDBC columns are 1-based + return columns.get(column - 1); + } + + @Override + public int getColumnType(int column) throws SQLException { + Object value = rows.get(0).get(getColumnName(column)); + if (value instanceof String) return Types.VARCHAR; + if (value instanceof Integer) return Types.INTEGER; + if (value instanceof byte[]) return Types.BLOB; + // add more types as needed + return Types.OTHER; + } + + // Implement other methods as needed + @Override public boolean isAutoIncrement(int column) { return false; } + @Override public boolean isCaseSensitive(int column) { return true; } + @Override public boolean isSearchable(int column) { return true; } + @Override public boolean isCurrency(int column) { return false; } + @Override public int isNullable(int column) { return ResultSetMetaData.columnNullable; } + @Override public boolean isSigned(int column) { return true; } + @Override public int getColumnDisplaySize(int column) { return 255; } + @Override public String getColumnLabel(int column) throws SQLException { return getColumnName(column); } + @Override public String getSchemaName(int column) { return ""; } + @Override public int getPrecision(int column) { return 0; } + @Override public int getScale(int column) { return 0; } + @Override public String getTableName(int column) { return ""; } + @Override public String getCatalogName(int column) { return ""; } + @Override public String getColumnTypeName(int column) throws SQLException { return getColumnName(column); } + @Override public boolean isReadOnly(int column) { return true; } + @Override public boolean isWritable(int column) { return false; } + @Override public boolean isDefinitelyWritable(int column) { return false; } + @Override public String getColumnClassName(int column) { return Object.class.getName(); } + @Override public T unwrap(Class iface) { return null; } + @Override public boolean isWrapperFor(Class iface) { return false; } + }; + } + + @Override + public boolean next() throws SQLException { + if ((rows != null && cursor + 1 < rows.size()) || (rowsByIndex != null && cursor + 1 < rowsByIndex.size())) { + cursor++; + return true; + } + return false; + } + + @Override + public void close() throws SQLException { + + } + + @Override + public boolean wasNull() throws SQLException { + return false; + } + + @Override + public String getString(int columnIndex) throws SQLException { + return null; + } + + @Override + public boolean getBoolean(int columnIndex) throws SQLException { + return false; + } + + @Override + public byte getByte(int columnIndex) throws SQLException { + return 0; + } + + @Override + public short getShort(int columnIndex) throws SQLException { + return 0; + } + + @Override + public int getInt(int columnIndex) throws SQLException { + return 0; + } + + @Override + public long getLong(int columnIndex) throws SQLException { + return 0; + } + + @Override + public float getFloat(int columnIndex) throws SQLException { + return 0; + } + + @Override + public double getDouble(int columnIndex) throws SQLException { + return 0; + } + + @Override + public BigDecimal getBigDecimal(int columnIndex, int scale) throws SQLException { + return null; + } + + @Override + public byte[] getBytes(int columnIndex) throws SQLException { + return new byte[0]; + } + + @Override + public Date getDate(int columnIndex) throws SQLException { + return null; + } + + @Override + public Time getTime(int columnIndex) throws SQLException { + return null; + } + + @Override + public Timestamp getTimestamp(int columnIndex) throws SQLException { + return null; + } + + @Override + public InputStream getAsciiStream(int columnIndex) throws SQLException { + return null; + } + + @Override + public InputStream getUnicodeStream(int columnIndex) throws SQLException { + return null; + } + + @Override + public InputStream getBinaryStream(int columnIndex) throws SQLException { + return ((Blob) rowsByIndex.get(cursor).get(columnIndex - 1)).getBinaryStream(); + } + + @Override + public String getString(String columnLabel) throws SQLException { + return (String) rows.get(cursor).get(columnLabel); + } + + @Override + public boolean getBoolean(String columnLabel) throws SQLException { + return false; + } + + @Override + public byte getByte(String columnLabel) throws SQLException { + return 0; + } + + @Override + public short getShort(String columnLabel) throws SQLException { + return 0; + } + + @Override + public int getInt(String columnLabel) throws SQLException { + return (Integer) rows.get(cursor).get(columnLabel); + } + + @Override + public long getLong(String columnLabel) throws SQLException { + return 0; + } + + @Override + public float getFloat(String columnLabel) throws SQLException { + return 0; + } + + @Override + public double getDouble(String columnLabel) throws SQLException { + return 0; + } + + @Override + public BigDecimal getBigDecimal(String columnLabel, int scale) throws SQLException { + return null; + } + + @Override + public byte[] getBytes(String columnLabel) throws SQLException { + return new byte[0]; + } + + @Override + public Date getDate(String columnLabel) throws SQLException { + return null; + } + + @Override + public Time getTime(String columnLabel) throws SQLException { + return null; + } + + @Override + public Timestamp getTimestamp(String columnLabel) throws SQLException { + return null; + } + + @Override + public InputStream getAsciiStream(String columnLabel) throws SQLException { + return null; + } + + @Override + public InputStream getUnicodeStream(String columnLabel) throws SQLException { + return null; + } + + @Override + public InputStream getBinaryStream(String columnLabel) throws SQLException { + return null; + } + + @Override + public SQLWarning getWarnings() throws SQLException { + return null; + } + + @Override + public void clearWarnings() throws SQLException { + + } + + @Override + public String getCursorName() throws SQLException { + return null; + } + + + @Override + public Object getObject(int columnIndex) throws SQLException { + return rowsByIndex.get(cursor).get(columnIndex - 1); + } + + @Override + public Object getObject(String columnLabel) throws SQLException { + return null; + } + + @Override + public int findColumn(String columnLabel) throws SQLException { + return 0; + } + + @Override + public Reader getCharacterStream(int columnIndex) throws SQLException { + return null; + } + + @Override + public Reader getCharacterStream(String columnLabel) throws SQLException { + return null; + } + + @Override + public BigDecimal getBigDecimal(int columnIndex) throws SQLException { + return null; + } + + @Override + public BigDecimal getBigDecimal(String columnLabel) throws SQLException { + return null; + } + + @Override + public boolean isBeforeFirst() throws SQLException { + return false; + } + + @Override + public boolean isAfterLast() throws SQLException { + return false; + } + + @Override + public boolean isFirst() throws SQLException { + return false; + } + + @Override + public boolean isLast() throws SQLException { + return false; + } + + @Override + public void beforeFirst() throws SQLException { + + } + + @Override + public void afterLast() throws SQLException { + + } + + @Override + public boolean first() throws SQLException { + return false; + } + + @Override + public boolean last() throws SQLException { + return false; + } + + @Override + public int getRow() throws SQLException { + return 0; + } + + @Override + public boolean absolute(int row) throws SQLException { + return false; + } + + @Override + public boolean relative(int rows) throws SQLException { + return false; + } + + @Override + public boolean previous() throws SQLException { + return false; + } + + @Override + public void setFetchDirection(int direction) throws SQLException { + + } + + @Override + public int getFetchDirection() throws SQLException { + return 0; + } + + @Override + public void setFetchSize(int rows) throws SQLException { + + } + + @Override + public int getFetchSize() throws SQLException { + return 0; + } + + @Override + public int getType() throws SQLException { + return 0; + } + + @Override + public int getConcurrency() throws SQLException { + return 0; + } + + @Override + public boolean rowUpdated() throws SQLException { + return false; + } + + @Override + public boolean rowInserted() throws SQLException { + return false; + } + + @Override + public boolean rowDeleted() throws SQLException { + return false; + } + + @Override + public void updateNull(int columnIndex) throws SQLException { + + } + + @Override + public void updateBoolean(int columnIndex, boolean x) throws SQLException { + + } + + @Override + public void updateByte(int columnIndex, byte x) throws SQLException { + + } + + @Override + public void updateShort(int columnIndex, short x) throws SQLException { + + } + + @Override + public void updateInt(int columnIndex, int x) throws SQLException { + + } + + @Override + public void updateLong(int columnIndex, long x) throws SQLException { + + } + + @Override + public void updateFloat(int columnIndex, float x) throws SQLException { + + } + + @Override + public void updateDouble(int columnIndex, double x) throws SQLException { + + } + + @Override + public void updateBigDecimal(int columnIndex, BigDecimal x) throws SQLException { + + } + + @Override + public void updateString(int columnIndex, String x) throws SQLException { + + } + + @Override + public void updateBytes(int columnIndex, byte[] x) throws SQLException { + + } + + @Override + public void updateDate(int columnIndex, Date x) throws SQLException { + + } + + @Override + public void updateTime(int columnIndex, Time x) throws SQLException { + + } + + @Override + public void updateTimestamp(int columnIndex, Timestamp x) throws SQLException { + + } + + @Override + public void updateAsciiStream(int columnIndex, InputStream x, int length) throws SQLException { + + } + + @Override + public void updateBinaryStream(int columnIndex, InputStream x, int length) throws SQLException { + + } + + @Override + public void updateCharacterStream(int columnIndex, Reader x, int length) throws SQLException { + + } + + @Override + public void updateObject(int columnIndex, Object x, int scaleOrLength) throws SQLException { + + } + + @Override + public void updateObject(int columnIndex, Object x) throws SQLException { + + } + + @Override + public void updateNull(String columnLabel) throws SQLException { + + } + + @Override + public void updateBoolean(String columnLabel, boolean x) throws SQLException { + + } + + @Override + public void updateByte(String columnLabel, byte x) throws SQLException { + + } + + @Override + public void updateShort(String columnLabel, short x) throws SQLException { + + } + + @Override + public void updateInt(String columnLabel, int x) throws SQLException { + + } + + @Override + public void updateLong(String columnLabel, long x) throws SQLException { + + } + + @Override + public void updateFloat(String columnLabel, float x) throws SQLException { + + } + + @Override + public void updateDouble(String columnLabel, double x) throws SQLException { + + } + + @Override + public void updateBigDecimal(String columnLabel, BigDecimal x) throws SQLException { + + } + + @Override + public void updateString(String columnLabel, String x) throws SQLException { + + } + + @Override + public void updateBytes(String columnLabel, byte[] x) throws SQLException { + + } + + @Override + public void updateDate(String columnLabel, Date x) throws SQLException { + + } + + @Override + public void updateTime(String columnLabel, Time x) throws SQLException { + + } + + @Override + public void updateTimestamp(String columnLabel, Timestamp x) throws SQLException { + + } + + @Override + public void updateAsciiStream(String columnLabel, InputStream x, int length) throws SQLException { + + } + + @Override + public void updateBinaryStream(String columnLabel, InputStream x, int length) throws SQLException { + + } + + @Override + public void updateCharacterStream(String columnLabel, Reader reader, int length) throws SQLException { + + } + + @Override + public void updateObject(String columnLabel, Object x, int scaleOrLength) throws SQLException { + + } + + @Override + public void updateObject(String columnLabel, Object x) throws SQLException { + + } + + @Override + public void insertRow() throws SQLException { + + } + + @Override + public void updateRow() throws SQLException { + + } + + @Override + public void deleteRow() throws SQLException { + + } + + @Override + public void refreshRow() throws SQLException { + + } + + @Override + public void cancelRowUpdates() throws SQLException { + + } + + @Override + public void moveToInsertRow() throws SQLException { + + } + + @Override + public void moveToCurrentRow() throws SQLException { + + } + + @Override + public Statement getStatement() throws SQLException { + return null; + } + + @Override + public Object getObject(int columnIndex, Map> map) throws SQLException { + return null; + } + + @Override + public Ref getRef(int columnIndex) throws SQLException { + return null; + } + + @Override + public Blob getBlob(int columnIndex) throws SQLException { + return (Blob) rowsByIndex.get(cursor).get(columnIndex - 1); + } + + @Override + public Clob getClob(int columnIndex) throws SQLException { + return null; + } + + @Override + public Array getArray(int columnIndex) throws SQLException { + return null; + } + + @Override + public Object getObject(String columnLabel, Map> map) throws SQLException { + return null; + } + + @Override + public Ref getRef(String columnLabel) throws SQLException { + return null; + } + + @Override + public Blob getBlob(String columnLabel) throws SQLException { + return null; + } + + @Override + public Clob getClob(String columnLabel) throws SQLException { + return null; + } + + @Override + public Array getArray(String columnLabel) throws SQLException { + return null; + } + + @Override + public Date getDate(int columnIndex, Calendar cal) throws SQLException { + return null; + } + + @Override + public Date getDate(String columnLabel, Calendar cal) throws SQLException { + return null; + } + + @Override + public Time getTime(int columnIndex, Calendar cal) throws SQLException { + return null; + } + + @Override + public Time getTime(String columnLabel, Calendar cal) throws SQLException { + return null; + } + + @Override + public Timestamp getTimestamp(int columnIndex, Calendar cal) throws SQLException { + return null; + } + + @Override + public Timestamp getTimestamp(String columnLabel, Calendar cal) throws SQLException { + return null; + } + + @Override + public URL getURL(int columnIndex) throws SQLException { + return null; + } + + @Override + public URL getURL(String columnLabel) throws SQLException { + return null; + } + + @Override + public void updateRef(int columnIndex, Ref x) throws SQLException { + + } + + @Override + public void updateRef(String columnLabel, Ref x) throws SQLException { + + } + + @Override + public void updateBlob(int columnIndex, Blob x) throws SQLException { + + } + + @Override + public void updateBlob(String columnLabel, Blob x) throws SQLException { + + } + + @Override + public void updateClob(int columnIndex, Clob x) throws SQLException { + + } + + @Override + public void updateClob(String columnLabel, Clob x) throws SQLException { + + } + + @Override + public void updateArray(int columnIndex, Array x) throws SQLException { + + } + + @Override + public void updateArray(String columnLabel, Array x) throws SQLException { + + } + + @Override + public RowId getRowId(int columnIndex) throws SQLException { + return null; + } + + @Override + public RowId getRowId(String columnLabel) throws SQLException { + return null; + } + + @Override + public void updateRowId(int columnIndex, RowId x) throws SQLException { + + } + + @Override + public void updateRowId(String columnLabel, RowId x) throws SQLException { + + } + + @Override + public int getHoldability() throws SQLException { + return 0; + } + + @Override + public boolean isClosed() throws SQLException { + return false; + } + + @Override + public void updateNString(int columnIndex, String nString) throws SQLException { + + } + + @Override + public void updateNString(String columnLabel, String nString) throws SQLException { + + } + + @Override + public void updateNClob(int columnIndex, NClob nClob) throws SQLException { + + } + + @Override + public void updateNClob(String columnLabel, NClob nClob) throws SQLException { + + } + + @Override + public NClob getNClob(int columnIndex) throws SQLException { + return null; + } + + @Override + public NClob getNClob(String columnLabel) throws SQLException { + return null; + } + + @Override + public SQLXML getSQLXML(int columnIndex) throws SQLException { + return null; + } + + @Override + public SQLXML getSQLXML(String columnLabel) throws SQLException { + return null; + } + + @Override + public void updateSQLXML(int columnIndex, SQLXML xmlObject) throws SQLException { + + } + + @Override + public void updateSQLXML(String columnLabel, SQLXML xmlObject) throws SQLException { + + } + + @Override + public String getNString(int columnIndex) throws SQLException { + return null; + } + + @Override + public String getNString(String columnLabel) throws SQLException { + return null; + } + + @Override + public Reader getNCharacterStream(int columnIndex) throws SQLException { + return null; + } + + @Override + public Reader getNCharacterStream(String columnLabel) throws SQLException { + return null; + } + + @Override + public void updateNCharacterStream(int columnIndex, Reader x, long length) throws SQLException { + + } + + @Override + public void updateNCharacterStream(String columnLabel, Reader reader, long length) throws SQLException { + + } + + @Override + public void updateAsciiStream(int columnIndex, InputStream x, long length) throws SQLException { + + } + + @Override + public void updateBinaryStream(int columnIndex, InputStream x, long length) throws SQLException { + + } + + @Override + public void updateCharacterStream(int columnIndex, Reader x, long length) throws SQLException { + + } + + @Override + public void updateAsciiStream(String columnLabel, InputStream x, long length) throws SQLException { + + } + + @Override + public void updateBinaryStream(String columnLabel, InputStream x, long length) throws SQLException { + + } + + @Override + public void updateCharacterStream(String columnLabel, Reader reader, long length) throws SQLException { + + } + + @Override + public void updateBlob(int columnIndex, InputStream inputStream, long length) throws SQLException { + + } + + @Override + public void updateBlob(String columnLabel, InputStream inputStream, long length) throws SQLException { + + } + + @Override + public void updateClob(int columnIndex, Reader reader, long length) throws SQLException { + + } + + @Override + public void updateClob(String columnLabel, Reader reader, long length) throws SQLException { + + } + + @Override + public void updateNClob(int columnIndex, Reader reader, long length) throws SQLException { + + } + + @Override + public void updateNClob(String columnLabel, Reader reader, long length) throws SQLException { + + } + + @Override + public void updateNCharacterStream(int columnIndex, Reader x) throws SQLException { + + } + + @Override + public void updateNCharacterStream(String columnLabel, Reader reader) throws SQLException { + + } + + @Override + public void updateAsciiStream(int columnIndex, InputStream x) throws SQLException { + + } + + @Override + public void updateBinaryStream(int columnIndex, InputStream x) throws SQLException { + + } + + @Override + public void updateCharacterStream(int columnIndex, Reader x) throws SQLException { + + } + + @Override + public void updateAsciiStream(String columnLabel, InputStream x) throws SQLException { + + } + + @Override + public void updateBinaryStream(String columnLabel, InputStream x) throws SQLException { + + } + + @Override + public void updateCharacterStream(String columnLabel, Reader reader) throws SQLException { + + } + + @Override + public void updateBlob(int columnIndex, InputStream inputStream) throws SQLException { + + } + + @Override + public void updateBlob(String columnLabel, InputStream inputStream) throws SQLException { + + } + + @Override + public void updateClob(int columnIndex, Reader reader) throws SQLException { + + } + + @Override + public void updateClob(String columnLabel, Reader reader) throws SQLException { + + } + + @Override + public void updateNClob(int columnIndex, Reader reader) throws SQLException { + + } + + @Override + public void updateNClob(String columnLabel, Reader reader) throws SQLException { + + } + + @Override + public T getObject(int columnIndex, Class type) throws SQLException { + return null; + } + + @Override + public T getObject(String columnLabel, Class type) throws SQLException { + return null; + } + + @Override + public T unwrap(Class iface) throws SQLException { + return null; + } + + @Override + public boolean isWrapperFor(Class iface) throws SQLException { + return false; + } + +}