Fix Win32 launcher, usage.

Some more characters need to be espaced now that SparkLauncher is
running a batch script as a a child. Also some fixes for the handling
of usage errors.

Author: Marcelo Vanzin

bin/spark-submit

@@ -18,4 +18,11 @@
 #
 
 SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"
+
+usage() {
+  "$SPARK_HOME"/bin/spark-class org.apache.spark.deploy.SparkSubmit --help
+  exit $1
+}
+export -f usage
+
 exec "$SPARK_HOME"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"

bin/spark-submit.cmd

@@ -20,4 +20,15 @@ rem
 rem This is the entry point for running Spark submit. To avoid polluting the
 rem environment, it just launches a new cmd to do the real work.
 
-cmd /V /E /C %~dp0spark-class2.cmd org.apache.spark.deploy.SparkSubmit %*
+set CLASS=org.apache.spark.deploy.SparkSubmit
+call %~dp0spark-class2.cmd  %CLASS% %*
+set SPARK_ERROR_LEVEL=%ERRORLEVEL%
+if "%SPARK_LAUNCHER_USAGE_ERROR%"=="1" (
+  call :usage
+  exit /b 1
+)
+exit /b %SPARK_ERROR_LEVEL%
+
+:usage
+call %SPARK_HOME%\bin\spark-class2.cmd %CLASS% --help
+goto :eof

launcher/src/main/java/org/apache/spark/launcher/CommandBuilderUtils.java

@@ -231,16 +231,19 @@ static void checkState(boolean check, String msg, Object... args) {
 
   /**
    * Quote a command argument for a command to be run by a Windows batch script, if the argument
-   * needs quoting. Arguments only seem to need quotes in batch scripts if they have whitespace.
+   * needs quoting. Arguments only seem to need quotes in batch scripts if they have certain
+   * special characters, some of which need extra (and different) escaping.
    *
    *  For example:
-   *    original single argument: ab "cde" fgh
-   *    quoted: "ab ""cde"" fgh"
+   *    original single argument: ab="cde fgh"
+   *    quoted: "ab^=""cde fgh"""
    */
   static String quoteForBatchScript(String arg) {
+
     boolean needsQuotes = false;
     for (int i = 0; i < arg.length(); i++) {
-      if (Character.isWhitespace(arg.codePointAt(i)) || arg.codePointAt(i) == '"') {
+      int c = arg.codePointAt(i);
+      if (Character.isWhitespace(c) || c == '"' || c == '=') {
         needsQuotes = true;
         break;
       }
@@ -252,8 +255,17 @@ static String quoteForBatchScript(String arg) {
     quoted.append("\"");
     for (int i = 0; i < arg.length(); i++) {
       int cp = arg.codePointAt(i);
-      if (cp == '\"') {
-        quoted.append("\"");
+      switch (cp) {
+      case '"':
+        quoted.append('"');
+        break;
+
+      case '=':
+        quoted.append('^');
+        break;
+
+      default:
+        break;
       }
       quoted.appendCodePoint(cp);
     }

launcher/src/main/java/org/apache/spark/launcher/Main.java

@@ -147,9 +147,9 @@ private static class UsageLauncher extends AbstractCommandBuilder {
     @Override
     public List<String> buildCommand(Map<String, String> env) {
       if (isWindows()) {
-        return Arrays.asList("set SPARK_LAUNCHER_USAGE_ERROR=1");
+        return Arrays.asList("set", "SPARK_LAUNCHER_USAGE_ERROR=1");
       } else {
-        return Arrays.asList("usage 1");
+        return Arrays.asList("usage", "1");
       }
     }
 

launcher/src/main/java/org/apache/spark/launcher/SparkLauncher.java

@@ -257,9 +257,21 @@ public SparkLauncher setVerbose(boolean verbose) {
    */
   public Process launch() throws IOException {
     List<String> cmd = new ArrayList<String>();
-    cmd.add(join(File.separator, builder.getSparkHome(), "bin", "spark-submit"));
+    String script = isWindows() ? "spark-submit.cmd" : "spark-submit";
+    cmd.add(join(File.separator, builder.getSparkHome(), "bin", script));
     cmd.addAll(builder.buildSparkSubmitArgs());
 
+    // Since the child process is a batch script, let's quote things so that special characters are
+    // preserved, otherwise the batch interpreter will mess up the arguments. Batch scripts are
+    // weird.
+    if (isWindows()) {
+      List<String> winCmd = new ArrayList<String>();
+      for (String arg : cmd) {
+        winCmd.add(quoteForBatchScript(arg));
+      }
+      cmd = winCmd;
+    }
+
     ProcessBuilder pb = new ProcessBuilder(cmd.toArray(new String[cmd.size()]));
     for (Map.Entry<String, String> e : builder.childEnv.entrySet()) {
       pb.environment().put(e.getKey(), e.getValue());

launcher/src/main/java/org/apache/spark/launcher/SparkSubmitCommandBuilder.java

@@ -245,6 +245,7 @@ private List<String> buildPySparkShellCommand(Map<String, String> env) throws IO
 
   private boolean isClientMode(Properties userProps) {
     String userMaster = firstNonEmpty(master, (String) userProps.get(SparkLauncher.SPARK_MASTER));
+    // Default master is "local[*]", so assume client mode in that case.
     return userMaster == null ||
       "client".equals(deployMode) ||
       (!userMaster.equals("yarn-cluster") && deployMode == null);
@@ -292,7 +293,9 @@ protected boolean handle(String opt, String value) {
         }
       } else {
         sparkArgs.add(opt);
-        sparkArgs.add(value);
+        if (value != null) {
+          sparkArgs.add(value);
+        }
       }
       return true;
     }

launcher/src/test/java/org/apache/spark/launcher/CommandBuilderUtilsSuite.java

@@ -74,6 +74,7 @@ public void testWindowsBatchQuoting() {
     assertEquals("\"a b c\"", quoteForBatchScript("a b c"));
     assertEquals("\"a \"\"b\"\" c\"", quoteForBatchScript("a \"b\" c"));
     assertEquals("\"a\"\"b\"\"c\"", quoteForBatchScript("a\"b\"c"));
+    assertEquals("\"ab^=\"\"cd\"\"\"", quoteForBatchScript("ab=\"cd\""));
   }
 
   @Test