Set up build pipeline for Spark-3.0.0

azure-pipelines.yml

@@ -284,6 +284,15 @@ stages:
       env:
         SPARK_HOME: $(Build.BinariesDirectory)\spark-2.4.6-bin-hadoop2.7
 
+    - task: DotNetCoreCLI@2
+      displayName: 'E2E tests for Spark 3.0.0'
+      inputs:
+        command: test
+        projects: '**/Microsoft.Spark*.E2ETest/*.csproj'
+        arguments: '--configuration $(buildConfiguration)'
+      env:
+        SPARK_HOME: $(Build.BinariesDirectory)\spark-3.0.0-bin-hadoop2.7
+
 - stage: ForwardCompatibility
   displayName: E2E Forward Compatibility Tests
   dependsOn: Build
@@ -549,3 +558,12 @@ stages:
       env:
         SPARK_HOME: $(Build.BinariesDirectory)\spark-2.4.6-bin-hadoop2.7
 
+    - task: DotNetCoreCLI@2
+      displayName: 'E2E tests for Spark 3.0.0'
+      inputs:
+        command: test
+        projects: '**/Microsoft.Spark*.E2ETest/*.csproj'
+        arguments: '--configuration $(buildConfiguration) --filter $(TestsToFilterOut)'
+      env:
+        SPARK_HOME: $(Build.BinariesDirectory)\spark-3.0.0-bin-hadoop2.7
+

script/download-spark-distros.cmd

@@ -24,6 +24,6 @@ curl -k -L -o spark-2.4.3.tgz https://archive.apache.org/dist/spark/spark-2.4.3/
 curl -k -L -o spark-2.4.4.tgz https://archive.apache.org/dist/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz && tar xzvf spark-2.4.4.tgz
 curl -k -L -o spark-2.4.5.tgz https://archive.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz && tar xzvf spark-2.4.5.tgz
 curl -k -L -o spark-2.4.6.tgz https://archive.apache.org/dist/spark/spark-2.4.6/spark-2.4.6-bin-hadoop2.7.tgz && tar xzvf spark-2.4.6.tgz
+curl -k -L -o spark-3.0.0.tgz https://archive.apache.org/dist/spark/spark-3.0.0/spark-3.0.0-bin-hadoop2.7.tgz && tar xzvf spark-3.0.0.tgz
 
 endlocal
-

src/csharp/Extensions/Microsoft.Spark.Extensions.Delta.E2ETest/DeltaFixture.cs

@@ -14,9 +14,17 @@ public class DeltaFixture
 
         public DeltaFixture()
         {
+            Version sparkVersion = SparkSettings.Version;
+            string deltaVersion = sparkVersion.Major switch
+            {
+                2 => "delta-core_2.11:0.6.1",
+                3 => "delta-core_2.12:0.7.0",
+                _ => throw new NotSupportedException($"Spark {sparkVersion} not supported.")
+            };
+
             Environment.SetEnvironmentVariable(
                 SparkFixture.EnvironmentVariableNames.ExtraSparkSubmitArgs,
-                "--packages io.delta:delta-core_2.11:0.6.1 " +
+                $"--packages io.delta:{deltaVersion} " +
                 "--conf spark.databricks.delta.snapshotPartitions=2 " +
                 "--conf spark.sql.sources.parallelPartitionDiscovery.parallelism=5");
             SparkFixture = new SparkFixture();

src/csharp/Extensions/Microsoft.Spark.Extensions.Delta.E2ETest/DeltaTableTests.cs

@@ -30,7 +30,7 @@ public DeltaTableTests(DeltaFixture fixture)
         /// Run the end-to-end scenario from the Delta Quickstart tutorial.
         /// </summary>
         /// <see cref="https://docs.delta.io/latest/quick-start.html"/>
-        [SkipIfSparkVersionIsLessThan(Versions.V2_4_2)]
+        [SkipIfSparkVersionIsNotInRange(Versions.V2_4_2, Versions.V3_0_0)]
         public void TestTutorialScenario()
         {
             using var tempDirectory = new TemporaryDirectory();
@@ -223,7 +223,7 @@ void testWrapper(
         /// <summary>
         /// Test that methods return the expected signature.
         /// </summary>
-        [SkipIfSparkVersionIsLessThan(Versions.V2_4_2)]
+        [SkipIfSparkVersionIsNotInRange(Versions.V2_4_2, Versions.V3_0_0)]
         public void TestSignatures()
         {
             using var tempDirectory = new TemporaryDirectory();

src/csharp/Extensions/Microsoft.Spark.Extensions.Hyperspace.E2ETest/HyperspaceFixture.cs

@@ -12,9 +12,17 @@ public class HyperspaceFixture
     {
         public HyperspaceFixture()
         {
+            Version sparkVersion = SparkSettings.Version;
+            string hyperspaceVersion = sparkVersion.Major switch
+            {
+                2 => "hyperspace-core_2.11:0.2.0",
+                3 => "hyperspace-core_2.12:0.2.0",
+                _ => throw new NotSupportedException($"Spark {sparkVersion} not supported.")
+            };
+
             Environment.SetEnvironmentVariable(
                 SparkFixture.EnvironmentVariableNames.ExtraSparkSubmitArgs,
-                "--packages com.microsoft.hyperspace:hyperspace-core_2.11:0.1.0");
+                $"--packages com.microsoft.hyperspace:{hyperspaceVersion}");
 
             SparkFixture = new SparkFixture();
         }

src/csharp/Extensions/Microsoft.Spark.Extensions.Hyperspace.E2ETest/HyperspaceTests.cs

@@ -53,7 +53,7 @@ public void Dispose()
         /// <summary>
         /// Test the method signatures for all Hyperspace APIs.
         /// </summary>
-        [SkipIfSparkVersionIsLessThan(Versions.V2_4_0)]
+        [SkipIfSparkVersionIsNotInRange(Versions.V2_4_0, Versions.V3_0_0)]
         public void TestSignatures()
         {
             // Indexes API.
@@ -86,7 +86,7 @@ public void TestSignatures()
         /// <summary>
         /// Test E2E functionality of index CRUD APIs.
         /// </summary>
-        [SkipIfSparkVersionIsLessThan(Versions.V2_4_0)]
+        [SkipIfSparkVersionIsNotInRange(Versions.V2_4_0, Versions.V3_0_0)]
         public void TestIndexCreateAndDelete()
         {
             // Should be one active index.
@@ -116,7 +116,7 @@ public void TestIndexCreateAndDelete()
         /// <summary>
         /// Test that the explain API generates the expected string.
         /// </summary>
-        [SkipIfSparkVersionIsLessThan(Versions.V2_4_0)]
+        [SkipIfSparkVersionIsNotInRange(Versions.V2_4_0, Versions.V3_0_0)]
         public void TestExplainAPI()
         {
             // Run a query that hits the index.

src/csharp/Microsoft.Spark.E2ETest/IpcTests/BroadcastTests.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Linq;
+using Microsoft.Spark.E2ETest.Utils;
 using Microsoft.Spark.Sql;
 using Xunit;
 using static Microsoft.Spark.Sql.Functions;
@@ -34,7 +35,7 @@ public BroadcastTests(SparkFixture fixture)
         /// <summary>
         /// Test Broadcast support by using multiple broadcast variables in a UDF.
         /// </summary>
-        [Fact]
+        [SkipIfSparkVersionIsGreaterOrEqualTo(Versions.V3_0_0)]
         public void TestMultipleBroadcastWithoutEncryption()
         {
             var obj1 = new TestBroadcastVariable(1, "first");
@@ -55,7 +56,7 @@ public void TestMultipleBroadcastWithoutEncryption()
         /// Test Broadcast.Destroy() that destroys all data and metadata related to the broadcast
         /// variable and makes it inaccessible from workers.
         /// </summary>
-        [Fact]
+        [SkipIfSparkVersionIsGreaterOrEqualTo(Versions.V3_0_0)]
         public void TestDestroy()
         {
             var obj1 = new TestBroadcastVariable(5, "destroy");
@@ -96,7 +97,7 @@ public void TestDestroy()
         /// Test Broadcast.Unpersist() deletes cached copies of the broadcast on the executors. If
         /// the broadcast is used after unpersist is called, it is re-sent to the executors.
         /// </summary>
-        [Fact]
+        [SkipIfSparkVersionIsGreaterOrEqualTo(Versions.V3_0_0)]
         public void TestUnpersist()
         {
             var obj = new TestBroadcastVariable(1, "unpersist");

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

@@ -156,7 +156,7 @@ public void TestUDF()
             }
         }
 
-        [Fact]
+        [SkipIfSparkVersionIsGreaterOrEqualTo(Versions.V3_0_0)]
         public void TestVectorUdf()
         {
             Func<Int32Array, StringArray, StringArray> udf1Func =
@@ -224,7 +224,7 @@ public void TestVectorUdf()
             }
         }
 
-        [Fact]
+        [SkipIfSparkVersionIsGreaterOrEqualTo(Versions.V3_0_0)]
         public void TestDataFrameVectorUdf()
         {
             Func<Int32DataFrameColumn, ArrowStringDataFrameColumn, ArrowStringDataFrameColumn> udf1Func =
@@ -290,7 +290,7 @@ public void TestDataFrameVectorUdf()
             }
         }
 
-        [Fact]
+        [SkipIfSparkVersionIsGreaterOrEqualTo(Versions.V3_0_0)]
         public void TestGroupedMapUdf()
         {
             DataFrame df = _spark
@@ -369,7 +369,7 @@ private static RecordBatch ArrowBasedCountCharacters(RecordBatch records)
         }
 
 
-        [Fact]
+        [SkipIfSparkVersionIsGreaterOrEqualTo(Versions.V3_0_0)]
         public void TestDataFrameGroupedMapUdf()
         {
             DataFrame df = _spark

src/csharp/Microsoft.Spark.E2ETest/SparkSettings.cs

@@ -35,7 +35,13 @@ private static void InitVersion()
             // Spark 2.3.2 built for Hadoop 2.7.3
             string firstLine =
                 File.ReadLines($"{SparkHome}{Path.DirectorySeparatorChar}RELEASE").First();
-            Version = new Version(firstLine.Split(' ')[1]);
+
+            // Grab "2.3.2" from "Spark 2.3.2 built for Hadoop 2.7.3"
+            string versionStr = firstLine.Split(' ')[1];
+
+            // Strip anything below version number.
+            // For example, "3.0.0-preview" should become "3.0.0".
+            Version = new Version(versionStr.Split('-')[0]);
         }
     }
 }

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

@@ -13,7 +13,32 @@ public SkipIfSparkVersionIsLessThan(string version)
         {
             if (SparkSettings.Version < new Version(version))
             {
-                Skip = $"Ignore on Spark version ({SparkSettings.Version}) <= {version}";
+                Skip = $"Ignore on Spark version ({SparkSettings.Version}) < {version}";
+            }
+        }
+    }
+
+    public sealed class SkipIfSparkVersionIsGreaterOrEqualTo : FactAttribute
+    {
+        public SkipIfSparkVersionIsGreaterOrEqualTo(string version)
+        {
+            if (SparkSettings.Version >= new Version(version))
+            {
+                Skip = $"Ignore on Spark version ({SparkSettings.Version}) >= {version}";
+            }
+        }
+    }
+
+    // Skip if the spark version is not in range [minVersion, maxVersion).
+    public sealed class SkipIfSparkVersionIsNotInRange : FactAttribute
+    {
+        public SkipIfSparkVersionIsNotInRange(string minInclusive, string maxExclusive)
+        {
+            if (SparkSettings.Version < new Version(minInclusive) ||
+                SparkSettings.Version >= new Version(maxExclusive))
+            {
+                Skip = $"Ignore on Spark version ({SparkSettings.Version}) not in range of " +
+                    $"[{minInclusive}, {maxExclusive})";
             }
         }
     }

src/scala/microsoft-spark-3.0.x/pom.xml

@@ -12,7 +12,7 @@
     <encoding>UTF-8</encoding>
     <scala.version>2.12.8</scala.version>
     <scala.binary.version>2.12</scala.binary.version>
-    <spark.version>3.0.0-preview</spark.version>
+    <spark.version>3.0.0</spark.version>
   </properties>
 
   <dependencies>

src/scala/pom.xml

@@ -13,6 +13,7 @@
   <modules>
     <module>microsoft-spark-2.3.x</module>
     <module>microsoft-spark-2.4.x</module>
+    <module>microsoft-spark-3.0.x</module>
   </modules>
 
   <pluginRepositories>