[DELTA-OSS-EXTERNAL] Added SQL examples and integration tests for python/scala

- Modified old tests to work with Spark 3.0 by adding the right spark conf
- Added a new test to test SQL on metastore tables and paths
- Updated integration testing script to run the examples in a fine-grained manner.

Originally done by @rahulsmahadev in #426

Closes #451

Co-authored-by: Rahul Mahadev <[email protected]>
Signed-off-by: Tathagata Das <[email protected]>

Author: Tathagata Das <[email protected]>

#10659 is resolved by tdas/qfwc2lq2.

GitOrigin-RevId: 4247f17fc78e741b01f5ee47cc8b96f776f4b250

Author: tdas

examples/README.md

@@ -2,9 +2,9 @@
 In this folder there are examples taken from the delta.io quickstart guide and docs. They are available in both Scala and Python and can be run if the prerequisites are satisfied.
 
 ### Prerequisites
-* Apache Spark version 2.4.2 or above
-* PySpark is required for running python examples
+* For Python examples, PySpark 3.0.0 or above needs to be installed.
+* For Scala examples, Spark does not need to be install because it depends on Spark maven artifacts.
 
 ### Instructions
-* To run an example in Python run `spark-submit --packages io.delta:delta-core_2.11:0.5.0 PATH/TO/EXAMPLE`
-* To run the Scala examples, `cd examples/scala` and run `./build/sbt "runMain example.{Example class name}"` e.g. `./build/sbt "runMain example.Quickstart"`
+* To run an example in Python run `spark-submit --packages io.delta:delta-core_2.12:0.7.0 PATH/TO/EXAMPLE`
+* To run the Scala examples, `cd examples/scala` and run `./build/sbt "runMain example.{Example class name}"` e.g. `./build/sbt "runMain example.Quickstart"`

examples/python/quickstart.py

@@ -14,8 +14,7 @@
 # limitations under the License.
 #
 
-from pyspark import SparkContext
-from pyspark.sql import Column, DataFrame, SparkSession, SQLContext, functions
+from pyspark.sql import Column, DataFrame, SparkSession, functions
 from pyspark.sql.functions import *
 from py4j.java_collections import MapConverter
 from delta.tables import *
@@ -28,14 +27,13 @@
 except:
     pass
 
-# Create SparkContext
-sc = SparkContext()
-sqlContext = SQLContext(sc)
-
-spark = SparkSession \
-    .builder \
+# Enable SQL commands and Update/Delete/Merge for the current spark session.
+# we need to set the following configs
+spark = SparkSession.builder \
     .appName("quickstart") \
     .master("local[*]") \
+    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
+    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
     .getOrCreate()
 
 # Create a table

examples/python/quickstart_sql.py

@@ -0,0 +1,73 @@
+
+from pyspark.sql import Column, DataFrame, SparkSession, functions
+from pyspark.sql.functions import *
+from py4j.java_collections import MapConverter
+from delta.tables import *
+import shutil
+import threading
+
+tableName = "tbltestpython"
+
+# Enable SQL/DML commands and Metastore tables for the current spark session.
+# We need to set the following configs
+
+spark = SparkSession.builder \
+    .appName("quickstart_sql") \
+    .master("local[*]") \
+    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
+    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
+    .getOrCreate()
+
+# Clear any previous runs
+spark.sql("DROP TABLE IF EXISTS " + tableName)
+spark.sql("DROP TABLE IF EXISTS newData")
+
+try:
+    # Create a table
+    print("############# Creating a table ###############")
+    spark.sql("CREATE TABLE %s(id LONG) USING delta" % tableName)
+    spark.sql("INSERT INTO %s VALUES 0, 1, 2, 3, 4" % tableName)
+
+    # Read the table
+    print("############ Reading the table ###############")
+    spark.sql("SELECT * FROM %s" % tableName).show()
+
+    # Upsert (merge) new data
+    print("########### Upsert new data #############")
+    spark.sql("CREATE TABLE newData(id LONG) USING parquet")
+    spark.sql("INSERT INTO newData VALUES 3, 4, 5, 6")
+
+    spark.sql('''MERGE INTO {0} USING newData
+            ON {0}.id = newData.id
+            WHEN MATCHED THEN
+              UPDATE SET {0}.id = newData.id
+            WHEN NOT MATCHED THEN INSERT *
+        '''.format(tableName))
+
+    spark.sql("SELECT * FROM %s" % tableName).show()
+
+    # Update table data
+    print("########## Overwrite the table ###########")
+    spark.sql("INSERT OVERWRITE %s select * FROM (VALUES 5, 6, 7, 8, 9) x (id)" % tableName)
+    spark.sql("SELECT * FROM %s" % tableName).show()
+
+    # Update every even value by adding 100 to it
+    print("########### Update to the table(add 100 to every even value) ##############")
+    spark.sql("UPDATE {0} SET id = (id + 100) WHERE (id % 2 == 0)".format(tableName))
+    spark.sql("SELECT * FROM %s" % tableName).show()
+
+    # Delete every even value
+    print("######### Delete every even value ##############")
+    spark.sql("DELETE FROM {0} WHERE (id % 2 == 0)".format(tableName))
+    spark.sql("SELECT * FROM %s" % tableName).show()
+
+    # Read old version of data using time travel
+    print("######## Read old data using time travel ############")
+    df = spark.read.format("delta").option("versionAsOf", 0).table(tableName)
+    df.show()
+
+finally:
+    # cleanup
+    spark.sql("DROP TABLE " + tableName)
+    spark.sql("DROP TABLE IF EXISTS newData")
+    spark.stop()

examples/python/quickstart_sql_on_paths.py

@@ -0,0 +1,73 @@
+
+from pyspark.sql import Column, DataFrame, SparkSession, functions
+from pyspark.sql.functions import *
+from py4j.java_collections import MapConverter
+from delta.tables import *
+import shutil
+import threading
+import tempfile
+import os
+
+table_dir = "/tmp/delta-table"
+# Clear any previous runs
+try:
+    shutil.rmtree(table_dir)
+except:
+    pass
+
+# Enable SQL/DML commands and Metastore tables for the current spark session.
+# We need to set the following configs
+
+spark = SparkSession.builder \
+    .appName("quickstart_sql_on_paths") \
+    .master("local[*]") \
+    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
+    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
+    .getOrCreate()
+
+# Clear any previous runs
+spark.sql("DROP TABLE IF EXISTS newData")
+
+try:
+    # Create a table
+    print("############# Creating a table ###############")
+    spark.sql("CREATE TABLE delta.`%s`(id LONG) USING delta" % table_dir)
+    spark.sql("INSERT INTO delta.`%s` VALUES 0, 1, 2, 3, 4" % table_dir)
+
+    # Read the table
+    print("############ Reading the table ###############")
+    spark.sql("SELECT * FROM delta.`%s`" % table_dir).show()
+
+    # Upsert (merge) new data
+    print("########### Upsert new data #############")
+    spark.sql("CREATE TABLE newData(id LONG) USING parquet")
+    spark.sql("INSERT INTO newData VALUES 3, 4, 5, 6")
+
+    spark.sql('''MERGE INTO delta.`{0}` AS data USING newData
+            ON data.id = newData.id
+            WHEN MATCHED THEN
+              UPDATE SET data.id = newData.id
+            WHEN NOT MATCHED THEN INSERT *
+        '''.format(table_dir))
+
+    spark.sql("SELECT * FROM delta.`%s`" % table_dir).show()
+
+    # Update table data
+    print("########## Overwrite the table ###########")
+    spark.sql("INSERT OVERWRITE delta.`%s` select * FROM (VALUES 5, 6, 7, 8, 9) x (id)" % table_dir)
+    spark.sql("SELECT * FROM delta.`%s`" % table_dir).show()
+
+    # Update every even value by adding 100 to it
+    print("########### Update to the table(add 100 to every even value) ##############")
+    spark.sql("UPDATE delta.`{0}` SET id = (id + 100) WHERE (id % 2 == 0)".format(table_dir))
+    spark.sql("SELECT * FROM delta.`%s`" % table_dir).show()
+
+    # Delete every even value
+    print("######### Delete every even value ##############")
+    spark.sql("DELETE FROM delta.`{0}` WHERE (id % 2 == 0)".format(table_dir))
+    spark.sql("SELECT * FROM delta.`%s`" % table_dir).show()
+
+finally:
+    # cleanup
+    spark.sql("DROP TABLE IF EXISTS newData")
+    spark.stop()

examples/python/streaming.py

@@ -14,52 +14,47 @@
 # limitations under the License.
 #
 
-from pyspark import SparkContext
-from pyspark.sql import Column, DataFrame, SparkSession, SQLContext, functions
+from pyspark.sql import Column, DataFrame, SparkSession, functions
 from pyspark.sql.functions import *
 from py4j.java_collections import MapConverter
 from delta.tables import *
 import shutil
 import random
 import threading
 
-# Clear previous run delta-tables
-files = ["/tmp/delta-table", "/tmp/delta-table2", "/tmp/delta-table3", "/tmp/delta-table4",
-         "/tmp/delta-table5", "/tmp/checkpoint/tbl1"]
-for i in files:
-    try:
-        shutil.rmtree(i)
-    except:
-        pass
-
-# Create SparkContext
-sc = SparkContext()
-sqlContext = SQLContext(sc)
-
-spark = SparkSession \
-    .builder \
+
+# Enable SQL commands and Update/Delete/Merge for the current spark session.
+# we need to set the following configs
+spark = SparkSession.builder \
     .appName("streaming") \
     .master("local[*]") \
+    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
+    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
     .getOrCreate()
 
+try:
+    shutil.rmtree("/tmp/delta-streaming/")
+except:
+    pass
+
 # Create a table(key, value) of some data
 data = spark.range(8)
 data = data.withColumn("value", data.id + random.randint(0, 5000))
-data.write.format("delta").save("/tmp/delta-table")
+data.write.format("delta").save("/tmp/delta-streaming/delta-table")
 
 # Stream writes to the table
 print("####### Streaming write ######")
 streamingDf = spark.readStream.format("rate").load()
 stream = streamingDf.selectExpr("value as id").writeStream\
     .format("delta")\
-    .option("checkpointLocation", "/tmp/checkpoint")\
-    .start("/tmp/delta-table2")
+    .option("checkpointLocation", "/tmp/delta-streaming/checkpoint")\
+    .start("/tmp/delta-streaming/delta-table2")
 stream.awaitTermination(10)
 stream.stop()
 
 # Stream reads from a table
 print("##### Reading from stream ######")
-stream2 = spark.readStream.format("delta").load("/tmp/delta-table2")\
+stream2 = spark.readStream.format("delta").load("/tmp/delta-streaming/delta-table2")\
     .writeStream\
     .format("console")\
     .start()
@@ -82,7 +77,7 @@ def upsertToDelta(microBatchOutputDF, batchId):
     .withColumn("id", col("value") % 10)\
     .drop("timestamp")
 # Write the output of a streaming aggregation query into Delta Lake table
-deltaTable = DeltaTable.forPath(spark, "/tmp/delta-table")
+deltaTable = DeltaTable.forPath(spark, "/tmp/delta-streaming/delta-table")
 print("#############  Original Delta Table ###############")
 deltaTable.toDF().show()
 stream3 = streamingAggregatesDF.writeStream\
@@ -98,8 +93,8 @@ def upsertToDelta(microBatchOutputDF, batchId):
 # Streaming append and concurrent repartition using  data change = false
 # tbl1 is the sink and tbl2 is the source
 print("############ Streaming appends with concurrent table repartition  ##########")
-tbl1 = "/tmp/delta-table4"
-tbl2 = "/tmp/delta-table5"
+tbl1 = "/tmp/delta-streaming/delta-table4"
+tbl2 = "/tmp/delta-streaming/delta-table5"
 numRows = 10
 spark.range(numRows).write.mode("overwrite").format("delta").save(tbl1)
 spark.read.format("delta").load(tbl1).show()
@@ -110,7 +105,7 @@ def upsertToDelta(microBatchOutputDF, batchId):
 # Prior to Delta 0.5.0 this would throw StreamingQueryException: Detected a data update in the
 # source table. This is currently not supported.
 stream4 = spark.readStream.format("delta").load(tbl2).writeStream.format("delta")\
-    .option("checkpointLocation", "/tmp/checkpoint/tbl1") \
+    .option("checkpointLocation", "/tmp/delta-streaming/checkpoint/tbl1") \
     .outputMode("append") \
     .start(tbl1)
 
@@ -126,8 +121,7 @@ def upsertToDelta(microBatchOutputDF, batchId):
 print("######### After streaming write #########")
 spark.read.format("delta").load(tbl1).show()
 # cleanup
-for i in files:
-    try:
-        shutil.rmtree(i)
-    except:
-        pass
+try:
+    shutil.rmtree("/tmp/delta-streaming/")
+except:
+    pass

examples/python/utilities.py

@@ -14,42 +14,26 @@
 # limitations under the License.
 #
 
-from pyspark import SparkContext
-from pyspark.sql import Column, DataFrame, SparkSession, SQLContext, functions
+from pyspark.sql import Column, DataFrame, SparkSession, functions
 from pyspark.sql.functions import *
 from py4j.java_collections import MapConverter
 from delta.tables import *
 import shutil
 import threading
 
-# Clear previous run's delta-tables
-try:
-    shutil.rmtree("/tmp/delta-table")
-except:
-    pass
-
-# Create SparkContext
-sc = SparkContext()
-sqlContext = SQLContext(sc)
-
-# Enable SQL for the current spark session. we need to set the following configs to enable SQL
-# Commands
-# config io.delta.sql.DeltaSparkSessionExtension -- to enable custom Delta-specific SQL commands
-# config parallelPartitionDiscovery.parallelism -- control the parallelism for vacuum
-spark = SparkSession \
-    .builder \
+spark = SparkSession.builder \
     .appName("utilities") \
     .master("local[*]") \
     .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
-    .config("spark.sql.sources.parallelPartitionDiscovery.parallelism", "8") \
+    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
+    .config("spark.sql.sources.parallelPartitionDiscovery.parallelism", "4") \
     .getOrCreate()
 
-# Apache Spark 2.4.x has a known issue (SPARK-25003) that requires explicit activation
-# of the extension and cloning of the session. This will unnecessary in Apache Spark 3.x.
-if spark.sparkContext.version < "3.":
-    spark.sparkContext._jvm.io.delta.sql.DeltaSparkSessionExtension() \
-        .apply(spark._jsparkSession.extensions())
-    spark = SparkSession(spark.sparkContext, spark._jsparkSession.cloneSession())
+# Clear previous run's delta-tables
+try:
+    shutil.rmtree("/tmp/delta-table")
+except:
+    pass
 
 # Create a table
 print("########### Create a Parquet table ##############")

examples/scala/build.sbt

@@ -17,7 +17,7 @@
 name := "example"
 organization := "com.example"
 organizationName := "example"
-scalaVersion := "2.11.12"
+scalaVersion := "2.12.10"
 version := "0.1.0"
 
 def getDeltaVersion(): String = {
@@ -27,13 +27,13 @@ def getDeltaVersion(): String = {
     println("Using Delta version " + version)
     version
   } else {
-    "0.5.0"
+    "0.7.0"
   }
 }
 
 lazy val root = (project in file("."))
   .settings(
     name := "hello-world",
     libraryDependencies += "io.delta" %% "delta-core" % getDeltaVersion(),
-    libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.4.3",
+    libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.0.0",
     resolvers += "Delta" at "https://dl.bintray.com/delta-io/delta/")