DataFrameWriter Option fails

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/DataFrameReaderTests.cs

@@ -0,0 +1,85 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using Microsoft.Spark.Sql;
+using Xunit;
+
+namespace Microsoft.Spark.E2ETest.IpcTests
+{
+    [Collection("Spark E2E Tests")]
+    public class DataFrameReaderTests
+    {
+        private readonly SparkSession _spark;
+
+        public DataFrameReaderTests(SparkFixture fixture)
+        {
+            _spark = fixture.Spark;
+        }
+
+        /// <summary>
+        /// Test signatures for APIs up to Spark 2.3.*.
+        /// </summary>
+        [Fact]
+        public void TestSignaturesV2_3_X()
+        {
+            DataFrameReader dfr = _spark.Read();
+
+            Assert.IsType<DataFrameReader>(dfr.Format("json"));
+
+            Assert.IsType<DataFrameReader>(dfr.Schema("age INT, name STRING"));
+
+            Assert.IsType<DataFrameReader>(dfr.Option("stringOption", "value"));
+            Assert.IsType<DataFrameReader>(dfr.Option("boolOption", true));
+            Assert.IsType<DataFrameReader>(dfr.Option("longOption", 1L));
+            Assert.IsType<DataFrameReader>(dfr.Option("doubleOption", 3D));
+
+            Assert.IsType<DataFrameReader>(
+                dfr.Options(
+                    new Dictionary<string, string>
+                    {
+                        { "option1", "value1" },
+                        { "option2", "value2" }
+                    }));
+
+            Assert.IsType<DataFrame>(dfr.Load(TestEnvironment.ResourceDirectory + "people.json"));
+            Assert.IsType<DataFrame>(
+                dfr.Load(
+                    TestEnvironment.ResourceDirectory + "people.csv",
+                    TestEnvironment.ResourceDirectory + "people.csv"));
+
+            Assert.IsType<DataFrame>(dfr.Json(TestEnvironment.ResourceDirectory + "people.json"));
+            Assert.IsType<DataFrame>(
+                dfr.Json(
+                    TestEnvironment.ResourceDirectory + "people.json",
+                    TestEnvironment.ResourceDirectory + "people.json"));
+
+            Assert.IsType<DataFrame>(dfr.Csv(TestEnvironment.ResourceDirectory + "people.csv"));
+            Assert.IsType<DataFrame>(
+                dfr.Csv(
+                    TestEnvironment.ResourceDirectory + "people.csv",
+                    TestEnvironment.ResourceDirectory + "people.csv"));
+
+            Assert.IsType<DataFrame>(
+                dfr.Parquet(TestEnvironment.ResourceDirectory + "users.parquet"));
+            Assert.IsType<DataFrame>(
+                dfr.Parquet(
+                    TestEnvironment.ResourceDirectory + "users.parquet",
+                    TestEnvironment.ResourceDirectory + "users.parquet"));
+
+            Assert.IsType<DataFrame>(dfr.Orc(TestEnvironment.ResourceDirectory + "users.orc"));
+            Assert.IsType<DataFrame>(
+                dfr.Orc(
+                    TestEnvironment.ResourceDirectory + "users.orc",
+                    TestEnvironment.ResourceDirectory + "users.orc"));
+
+            dfr = _spark.Read();
+            Assert.IsType<DataFrame>(dfr.Text(TestEnvironment.ResourceDirectory + "people.txt"));
+            Assert.IsType<DataFrame>(
+                dfr.Text(
+                    TestEnvironment.ResourceDirectory + "people.txt",
+                    TestEnvironment.ResourceDirectory + "people.txt"));
+        }
+    }
+}

src/csharp/Microsoft.Spark.E2ETest/IpcTests/Sql/DataFrameWriterTests.cs

@@ -0,0 +1,93 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Collections.Generic;
+using Microsoft.Spark.E2ETest.Utils;
+using Microsoft.Spark.Sql;
+using Xunit;
+
+namespace Microsoft.Spark.E2ETest.IpcTests
+{
+    [Collection("Spark E2E Tests")]
+    public class DataFrameWriterTests
+    {
+        private readonly SparkSession _spark;
+
+        public DataFrameWriterTests(SparkFixture fixture)
+        {
+            _spark = fixture.Spark;
+        }
+
+        /// <summary>
+        /// Test signatures for APIs up to Spark 2.3.*.
+        /// </summary>
+        [Fact]
+        public void TestSignaturesV2_3_X()
+        {
+            {
+                DataFrameWriter dfw = _spark
+                    .Read()
+                    .Schema("age INT, name STRING")
+                    .Json(TestEnvironment.ResourceDirectory + "people.json")
+                    .Write();
+
+                Assert.IsType<DataFrameWriter>(dfw.Mode(SaveMode.Ignore));
+
+                Assert.IsType<DataFrameWriter>(dfw.Mode("overwrite"));
+
+                Assert.IsType<DataFrameWriter>(dfw.Format("json"));
+
+                Assert.IsType<DataFrameWriter>(dfw.Option("stringOption", "value"));
+                Assert.IsType<DataFrameWriter>(dfw.Option("boolOption", true));
+                Assert.IsType<DataFrameWriter>(dfw.Option("longOption", 1L));
+                Assert.IsType<DataFrameWriter>(dfw.Option("doubleOption", 3D));
+
+                Assert.IsType<DataFrameWriter>(
+                    dfw.Options(
+                        new Dictionary<string, string>
+                        {
+                            { "option1", "value1" },
+                            { "option2", "value2" }
+                        }));
+
+
+                Assert.IsType<DataFrameWriter>(dfw.PartitionBy("age"));
+                Assert.IsType<DataFrameWriter>(dfw.PartitionBy("age", "name"));
+
+                Assert.IsType<DataFrameWriter>(dfw.BucketBy(3, "age"));
+                Assert.IsType<DataFrameWriter>(dfw.BucketBy(3, "age", "name"));
+
+                Assert.IsType<DataFrameWriter>(dfw.SortBy("name"));
+            }
+
+            using (var tempDir = new TemporaryDirectory(AppDomain.CurrentDomain.BaseDirectory))
+            {
+                DataFrameWriter dfw = _spark
+                    .Read()
+                    .Csv(TestEnvironment.ResourceDirectory + "people.csv")
+                    .Write();
+
+                // TODO: Test dfw.Jdbc without running a local db.
+
+                dfw.Option("path", tempDir.Path).SaveAsTable("TestTable");
+
+                dfw.InsertInto("TestTable");
+
+                dfw.Option("path", tempDir.Path + "TestSavePath1").Save();
+                dfw.Save(tempDir.Path + "TestSavePath2");
+
+                dfw.Json(tempDir.Path + "TestJsonPath");
+
+                dfw.Parquet(tempDir.Path + "TestParquetPath");
+
+                dfw.Orc(tempDir.Path + "TestOrcPath");
+
+                dfw.Text(tempDir.Path + "TestTextPath");
+
+                dfw.Csv(tempDir.Path + "TestCsvPath");
+            }
+        }
+    }
+}

src/csharp/Microsoft.Spark.E2ETest/Utils/TemporaryDirectory.cs

@@ -0,0 +1,64 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.IO;
+
+namespace Microsoft.Spark.E2ETest.Utils
+{
+    /// <summary>
+    /// Creates a temporary folder at a specified root path that is automatically
+    /// cleaned up when disposed.
+    /// </summary>
+    internal sealed class TemporaryDirectory : IDisposable
+    {
+        private bool disposed = false;
+
+        /// <summary>
+        /// Path to temporary folder.
+        /// </summary>
+        public string Path { get; }
+
+        public TemporaryDirectory(string rootDirectory)
+        {
+            Path = System.IO.Path.Combine(rootDirectory, Guid.NewGuid().ToString());
+            Cleanup();
+            Directory.CreateDirectory(Path);
+            Path += System.IO.Path.DirectorySeparatorChar;
+        }
+
+        public void Dispose()
+        {
+            // Dispose of unmanaged resources.
+            Dispose(true);
+            // Suppress finalization.
+            GC.SuppressFinalize(this);
+        }
+
+        private void Cleanup()
+        {
+            if (File.Exists(Path))
+            {
+                File.Delete(Path);
+            }
+            else if (Directory.Exists(Path))
+            {
+                Directory.Delete(Path, true);
+            }
+        }
+
+        private void Dispose(bool disposing)
+        {
+            if (disposed)
+                return;
+
+            if (disposing)
+            {
+                Cleanup();
+            }
+
+            disposed = true;
+        }
+    }
+}

src/csharp/Microsoft.Spark/Sql/DataFrameReader.cs

@@ -58,8 +58,7 @@ public DataFrameReader Schema(string schemaString)
         /// <returns>This DataFrameReader object</returns>
         public DataFrameReader Option(string key, string value)
         {
-            OptionInternal(key, value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -70,8 +69,7 @@ public DataFrameReader Option(string key, string value)
         /// <returns>This DataFrameReader object</returns>
         public DataFrameReader Option(string key, bool value)
         {
-            OptionInternal(key, value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -82,8 +80,7 @@ public DataFrameReader Option(string key, bool value)
         /// <returns>This DataFrameReader object</returns>
         public DataFrameReader Option(string key, long value)
         {
-            OptionInternal(key, value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -94,8 +91,7 @@ public DataFrameReader Option(string key, long value)
         /// <returns>This DataFrameReader object</returns>
         public DataFrameReader Option(string key, double value)
         {
-            OptionInternal(key, value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -118,8 +114,7 @@ public DataFrameReader Options(Dictionary<string, string> options)
         /// </remarks>
         /// <param name="paths">Input paths</param>
         /// <returns>DataFrame object</returns>
-        public DataFrame Load(params string[] paths) =>
-            new DataFrame((JvmObjectReference)_jvmObject.Invoke("load", paths));
+        public DataFrame Load(params string[] paths) => LoadSource("load", paths);
 
         /// <summary>
         /// Loads a JSON file (one object per line) and returns the result as a DataFrame.
@@ -182,7 +177,7 @@ private DataFrame LoadSource(string source, params string[] paths)
                 throw new ArgumentException($"paths cannot be empty for source: {source}");
             }
 
-            return new DataFrame((JvmObjectReference)_jvmObject.Invoke(source, paths));
+            return new DataFrame((JvmObjectReference)_jvmObject.Invoke(source, (object)paths));
         }
     }
 }

src/csharp/Microsoft.Spark/Sql/DataFrameWriter.cs

@@ -71,8 +71,7 @@ public DataFrameWriter Format(string source)
         /// <returns>This DataFrameWriter object</returns>
         public DataFrameWriter Option(string key, string value)
         {
-            _jvmObject.Invoke("option", value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -83,8 +82,7 @@ public DataFrameWriter Option(string key, string value)
         /// <returns>This DataFrameWriter object</returns>
         public DataFrameWriter Option(string key, bool value)
         {
-            _jvmObject.Invoke("option", value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -95,8 +93,7 @@ public DataFrameWriter Option(string key, bool value)
         /// <returns>This DataFrameWriter object</returns>
         public DataFrameWriter Option(string key, long value)
         {
-            _jvmObject.Invoke("option", value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -107,8 +104,7 @@ public DataFrameWriter Option(string key, long value)
         /// <returns>This DataFrameWriter object</returns>
         public DataFrameWriter Option(string key, double value)
         {
-            _jvmObject.Invoke("option", value);
-            return this;
+            return OptionInternal(key, value);
         }
 
         /// <summary>
@@ -240,6 +236,18 @@ public void Jdbc(string url, string table, Dictionary<string, string> properties
         /// Saves the content of the DataFrame in CSV format at the specified path.
         /// </summary>
         /// <param name="path">Path to save the content</param>
-        public void Csv(string path) => _jvmObject.Invoke("csvs", path);
+        public void Csv(string path) => _jvmObject.Invoke("csv", path);
+
+        /// <summary>
+        /// Helper function to add given key/value pair as a new option.
+        /// </summary>
+        /// <param name="key">Name of the option</param>
+        /// <param name="value">Value of the option</param>
+        /// <returns>This DataFrameWriter object</returns>
+        private DataFrameWriter OptionInternal(string key, object value)
+        {
+            _jvmObject.Invoke("option", key, value);
+            return this;
+        }
     }
 }

src/csharp/Microsoft.Spark/Sql/SparkSession.cs

@@ -4,7 +4,6 @@
 
 using System;
 using Microsoft.Spark.Interop.Ipc;
-using Microsoft.Spark.Services;
 using Microsoft.Spark.Sql.Streaming;
 
 namespace Microsoft.Spark.Sql