Add order Hints for Bulk Copy operations

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

@@ -38,6 +38,7 @@
 import java.util.Collections;
 import java.util.GregorianCalendar;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
@@ -111,6 +112,22 @@
         }
     }
 
+    private class ColumnOrderHint implements Serializable {
+        /**
+         * Always update serialVersionUID when prompted.
+         */
+        private static final long serialVersionUID = 6132627333120344137L;
+
+        String columnName = null;
+
+        SQLServerSortOrder sortOrder;
+
+        ColumnOrderHint(String columnName, SQLServerSortOrder sortOrder) {
+            this.columnName = columnName;
+            this.sortOrder = sortOrder;
+        }
+    }
+
     /**
      * Class name for logging.
      */
@@ -136,6 +153,11 @@
      */
     private List<ColumnMapping> columnMappings;
 
+    /**
+     * Column order hints describe the sort order of columns in the clustered index of the destination
+     */
+    private List<ColumnOrderHint> columnOrderHints;
+
     /**
      * Flag if SQLServerBulkCopy owns the connection and should close it when Close is called
      */
@@ -457,6 +479,43 @@
         loggerExternal.exiting(loggerClassName, "clearColumnMappings");
     }
 
+     /**
+     * Adds a new column order hint, specify the column name and sort order
+     * 
+     * @param columnName
+     *        Column name.
+     * @param sortOrder
+     *        Column sort order.
+     * @throws SQLServerException
+     *         If the column order hint is invalid
+     */
+    public void addColumnOrderHint(String columnName, SQLServerSortOrder sortOrder) throws SQLServerException {
+        if (loggerExternal.isLoggable(java.util.logging.Level.FINER)) {
+            loggerExternal.entering(loggerClassName, "addColumnOrderHint",
+                    new Object[] {columnName, sortOrder});
+        }
+
+        if (null == columnName || columnName.isEmpty()) {
+            throwInvalidArgument("columnName");
+        } else if (null == sortOrder || SQLServerSortOrder.UNSPECIFIED == sortOrder) {
+            throwInvalidArgument("sortOrder");
+        }
+        columnOrderHints.add(new ColumnOrderHint(columnName, sortOrder));
+
+        loggerExternal.exiting(loggerClassName, "addColumnOrderHint");
+    }
+
+     /**
+     * Clears the contents of the column order hints
+     */
+    public void clearColumnOrderHints() {
+        loggerExternal.entering(loggerClassName, "clearColumnOrderHints");
+
+        columnOrderHints.clear();
+
+        loggerExternal.exiting(loggerClassName, "clearColumnOrderHints");
+    }
+
     /**
      * Closes the SQLServerBulkCopy instance
      */
@@ -646,6 +705,7 @@
      */
     private void initializeDefaults() {
         columnMappings = new ArrayList<>();
+        columnOrderHints = new ArrayList<>();
         destinationTableName = null;
         serverBulkData = null;
         sourceResultSet = null;
@@ -1493,6 +1553,7 @@
     private String createInsertBulkCommand(TDSWriter tdsWriter) throws SQLServerException {
         StringBuilder bulkCmd = new StringBuilder();
         List<String> bulkOptions = new ArrayList<>();
+        Set<String> destColumns = new HashSet<>();
         String endColumn = " , ";
         bulkCmd.append("INSERT BULK ").append(destinationTableName).append(" (");
 
@@ -1501,8 +1562,12 @@
                 endColumn = " ) ";
             }
             ColumnMapping colMapping = columnMappings.get(i);
-            String columnCollation = destColumnMetadata
-                    .get(columnMappings.get(i).destinationColumnOrdinal).collationName;
+
+            BulkColumnMetaData columnMetaData = destColumnMetadata
+                    .get(columnMappings.get(i).destinationColumnOrdinal);
+            destColumns.add(columnMetaData.columnName);
+
+            String columnCollation = columnMetaData.collationName;
             String addCollate = "";
 
             String destType = getDestTypeFromSrcType(colMapping.sourceColumnOrdinal,
@@ -1547,6 +1612,36 @@
             bulkOptions.add("ALLOW_ENCRYPTED_VALUE_MODIFICATIONS");
         }
 
+        if (0 < columnOrderHints.size()) {
+            StringBuilder orderHintText = new StringBuilder("ORDER(");
+
+            for (ColumnOrderHint columnOrderHint : columnOrderHints) {
+                String columnName = columnOrderHint.columnName;
+
+                if (!destColumns.contains(columnName)) {
+                    MessageFormat form = new MessageFormat(
+                            SQLServerException.getErrString("R_invalidColumn"));
+                    Object[] msgArgs = { columnName };
+                    throw new SQLServerException(form.format(msgArgs), SQLState.COL_NOT_FOUND,
+                            DriverError.NOT_SET, null);
+                }
+
+                String sortOrderText = columnOrderHint.sortOrder == SQLServerSortOrder.DESCENDING ? "DESC" : "ASC";
+
+                if (columnName.contains("]")) {
+                    String escapedColumnName = columnName.replaceAll("]", "]]");
+                    orderHintText.append("[").append(escapedColumnName).append("] ").append(sortOrderText).append(", ");
+                } else {
+                    orderHintText.append("[").append(columnName).append("] ").append(sortOrderText).append(", ");
+                } 
+            }
+
+            orderHintText.setLength(orderHintText.length() - 2);
+            orderHintText.append(")");
+
+            bulkOptions.add(orderHintText.toString());
+        }
+
         Iterator<String> it = bulkOptions.iterator();
         if (it.hasNext()) {
             bulkCmd.append(" with (");