Add support for InputStream data type using setBinaryStream() for bulk copy for batch insert API (#2826)

* Added support for InputStream data type using setBinaryStream() for bulk copy for batch insert API

* Added proper comments

Author: muskan124947

src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkBatchInsertRecord.java

@@ -5,6 +5,7 @@
 
 package com.microsoft.sqlserver.jdbc;
 
+import java.io.InputStream;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.sql.Types;
@@ -132,10 +133,10 @@ private Object convertValue(ColumnMetadata cm, Object data) throws SQLServerExce
             case Types.VARBINARY:
             case Types.LONGVARBINARY:
             case Types.BLOB: {
-                if (data instanceof byte[]) {
+                if (data instanceof byte[] || data instanceof InputStream) {
                     /*
-                     * if the binary data comes in as a byte array through setBytes through Bulk Copy for Batch Insert
-                     * API, don't turn the binary array into a string.
+                     * if the binary data comes in as a byte array or Input Stream through setBytes()/setBinaryStream() 
+                     * through Bulk Copy for Batch Insert API, don't turn the binary array into a string.
                      */
                     return data;
                 } else {

src/test/java/com/microsoft/sqlserver/jdbc/preparedStatement/BatchExecutionWithBulkCopyTest.java

@@ -7,10 +7,13 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
 import java.sql.BatchUpdateException;
 import java.sql.Connection;
 import java.sql.Date;
@@ -1696,6 +1699,63 @@ private void getCreateTableWithStringData() throws SQLException {
         }
     }
 
+    /**
+     * Test bulk insert with InputStream data when useBulkCopyForBatchInsert is true.
+     * Added support for InputStream data type with bulk copy batch insert.
+     * 
+     * @throws Exception
+     */
+    @Test
+    public void testBulkCopyBatchInsertForInputStreamData() throws Exception {
+        String tableName = RandomUtil.getIdentifier("BulkTableForInputStream");
+        String insertSQL = "INSERT INTO " + AbstractSQLGenerator.escapeIdentifier(tableName) +
+                " (Id, Data) VALUES (?, ?)";
+        String selectSQL = "SELECT Id, Data FROM " + AbstractSQLGenerator.escapeIdentifier(tableName);
+
+        try (Connection connection = PrepUtil.getConnection(connectionString + ";useBulkCopyForBatchInsert=true;");
+                Statement stmt = connection.createStatement();
+                SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) connection.prepareStatement(insertSQL)) {
+
+            getCreateTableInputStream(tableName);
+
+            String data = "Sample string to be inserted as binary data.";
+            byte[] bytes = data.getBytes(StandardCharsets.UTF_8);
+            InputStream inputStream = new ByteArrayInputStream(bytes);
+
+            pstmt.setObject(1, 1);
+            pstmt.setBinaryStream(2, inputStream);
+            
+            pstmt.addBatch();
+            pstmt.executeBatch();
+
+            // Validate inserted data
+            try (ResultSet rs = stmt.executeQuery(selectSQL)) {
+                assertTrue(rs.next());
+
+                assertEquals(1, rs.getInt(1));
+                byte[] retrievedBytes = rs.getBytes(2);
+                String retrievedData = new String(retrievedBytes, StandardCharsets.UTF_8);
+                assertEquals(data, retrievedData);
+                
+            }
+        } finally {
+            try (Statement stmt = connection.createStatement()) {
+                TestUtils.dropTableIfExists(AbstractSQLGenerator.escapeIdentifier(tableName), stmt);
+            }
+        }
+    }
+
+    private void getCreateTableInputStream(String tableName) throws SQLException {
+        try (Statement stmt = connection.createStatement()) {
+            TestUtils.dropTableIfExists(AbstractSQLGenerator.escapeIdentifier(tableName), stmt);
+            String createTableSQL = "CREATE TABLE " + AbstractSQLGenerator.escapeIdentifier(tableName) + " (" +
+                    "Id INT PRIMARY KEY, " +
+                    "Data VARBINARY(MAX) NULL" + ")";
+
+            stmt.execute(createTableSQL);
+        }
+    }
+
     @BeforeAll
     public static void setupTests() throws Exception {
         setConnection();