Add Spark 3.3.0 support (#9)

Author: dazfuller

.github/workflows/main.yml

@@ -11,22 +11,22 @@ jobs:
   build:
 
     runs-on: ubuntu-latest
-    
+
     strategy:
       matrix:
-        spark-version: [3.0.1, 3.0.2, 3.0.3, 3.1.1, 3.1.2, 3.2.0]
+        spark-version: [3.0.1, 3.0.2, 3.0.3, 3.1.1, 3.1.2, 3.1.3, 3.2.0, 3.2.2, 3.3.0]
 
     steps:
     - uses: actions/checkout@v2
-    
+
     - name: Set up JDK 1.8
       uses: actions/setup-java@v1
       with:
         java-version: 1.8
-    
+
     - name: Test and package
       run: sbt -DsparkVersion="${{ matrix.spark-version }}" clean compile test package
-    
+
     - name: Upload the package
       uses: actions/upload-artifact@v2.2.0
       with:

build.sbt

@@ -30,8 +30,10 @@ libraryDependencies ++= Seq(
 Compile / unmanagedSourceDirectories ++= {
   if (sparkVersion.value < "3.2.0") {
     Seq(baseDirectory.value / "src/main/3.0/scala")
-  } else {
+  } else if (sparkVersion.value < "3.3.0") {
     Seq(baseDirectory.value / "src/main/3.2/scala")
+  } else {
+    Seq(baseDirectory.value / "src/main/3.3/scala")
   }
 }
 
@@ -46,13 +48,13 @@ libraryDependencies ++= Seq(
 
 // Define common settings for the library
 val commonSettings = Seq(
-  sparkVersion := System.getProperty("sparkVersion", "3.2.0"),
+  sparkVersion := System.getProperty("sparkVersion", "3.3.0"),
   scalaVersion := {
     if (sparkVersion.value >= "3.2.0") {
       "2.12.14"
     } else {
       "2.12.10"
     }
   },
-  scalaTestVersion := "3.2.10"
+  scalaTestVersion := "3.2.13"
 )

…/main/scala/com/bp/sds/cef/CefScan.scala …n/3.0/scala/com/bp/sds/cef/CefScan.scalasrc/main/scala/com/bp/sds/cef/CefScan.scala renamed to src/main/3.0/scala/com/bp/sds/cef/CefScan.scala src/main/scala/com/bp/sds/cef/CefScan.scala renamed to src/main/3.0/scala/com/bp/sds/cef/CefScan.scala

@@ -0,0 +1,38 @@
+package com.bp.sds.cef
+
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.read.PartitionReaderFactory
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.{FileScan, TextBasedFileScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.util.SerializableConfiguration
+
+import scala.collection.JavaConverters.mapAsScalaMapConverter
+
+private[cef] case class CefScan(
+                                 sparkSession: SparkSession,
+                                 fileIndex: PartitioningAwareFileIndex,
+                                 dataSchema: StructType,
+                                 readDataSchema: StructType,
+                                 readPartitionSchema: StructType,
+                                 options: CaseInsensitiveStringMap,
+                                 partitionFilters: Seq[Expression] = Seq.empty,
+                                 dataFilters: Seq[Expression] = Seq.empty
+                               ) extends TextBasedFileScan(sparkSession, options) {
+  private val optionsAsScala = options.asScala.toMap
+  private val cefOptions = CefParserOptions.from(options)
+
+  override def isSplitable(path: Path): Boolean = super.isSplitable(path)
+
+  override def withFilters(partitionFilters: Seq[Expression], dataFilters: Seq[Expression]): FileScan =
+    this.copy(partitionFilters = partitionFilters, dataFilters = dataFilters)
+
+  override def createReaderFactory(): PartitionReaderFactory = {
+    val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(optionsAsScala)
+    val broadcastConf = sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
+    CefPartitionReaderFactory(sparkSession.sessionState.conf, broadcastConf, dataSchema, readDataSchema, readPartitionSchema, cefOptions)
+  }
+}

src/main/3.2/scala/com/bp/sds/cef/CefScan.scala

@@ -0,0 +1,24 @@
+package com.bp.sds.cef
+
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.mapreduce.TaskAttemptContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.{CodecStreams, OutputWriter}
+import org.apache.spark.sql.types.StructType
+
+import java.nio.charset.StandardCharsets
+
+private[cef] class CefOutputWriter(val path: String, cefOptions: CefParserOptions, dataSchema: StructType, context: TaskAttemptContext) extends OutputWriter with Logging {
+  private val writer = CodecStreams.createOutputStreamWriter(context, new Path(path), StandardCharsets.UTF_8)
+  private val gen = CefRecordWriter(dataSchema, writer, cefOptions)
+
+  override def write(row: InternalRow): Unit = {
+    gen.writeRow(row)
+    gen.writeLineEnding()
+  }
+
+  override def close(): Unit = {
+    writer.close()
+  }
+}

src/main/3.3/scala/com/bp/sds/cef/CefOutputWriter.scala

@@ -0,0 +1,23 @@
+package com.bp.sds.cef
+
+import org.apache.hadoop.mapreduce.{Job, TaskAttemptContext}
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
+import org.apache.spark.sql.execution.datasources.{CodecStreams, OutputWriter, OutputWriterFactory}
+import org.apache.spark.sql.execution.datasources.v2.FileWrite
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, StructType}
+
+private[cef] case class CefOutputWriterBuilder(paths: Seq[String],
+                                               formatName: String,
+                                               supportsDataType: DataType => Boolean,
+                                               info: LogicalWriteInfo
+                                              ) extends FileWrite with WriteBuilder {
+  override def prepareWrite(sqlConf: SQLConf, job: Job, options: Map[String, String], dataSchema: StructType): OutputWriterFactory =
+    new OutputWriterFactory {
+      override def getFileExtension(context: TaskAttemptContext): String =
+        ".log" + CodecStreams.getCompressionExtension(context)
+
+      override def newInstance(path: String, dataSchema: StructType, context: TaskAttemptContext): OutputWriter =
+        new CefOutputWriter(path, CefParserOptions.from(options), dataSchema, context)
+    }
+}

src/main/3.3/scala/com/bp/sds/cef/CefOutputWriterBuilder.scala

@@ -0,0 +1,35 @@
+package com.bp.sds.cef
+
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.read.PartitionReaderFactory
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.{FileScan, TextBasedFileScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.util.SerializableConfiguration
+
+import scala.collection.JavaConverters.mapAsScalaMapConverter
+
+private[cef] case class CefScan(
+                                 sparkSession: SparkSession,
+                                 fileIndex: PartitioningAwareFileIndex,
+                                 dataSchema: StructType,
+                                 readDataSchema: StructType,
+                                 readPartitionSchema: StructType,
+                                 options: CaseInsensitiveStringMap,
+                                 partitionFilters: Seq[Expression] = Seq.empty,
+                                 dataFilters: Seq[Expression] = Seq.empty
+                               ) extends TextBasedFileScan(sparkSession, options) {
+  private val optionsAsScala = options.asScala.toMap
+  private val cefOptions = CefParserOptions.from(options)
+
+  override def isSplitable(path: Path): Boolean = super.isSplitable(path)
+
+  override def createReaderFactory(): PartitionReaderFactory = {
+    val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(optionsAsScala)
+    val broadcastConf = sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
+    CefPartitionReaderFactory(sparkSession.sessionState.conf, broadcastConf, dataSchema, readDataSchema, readPartitionSchema, cefOptions)
+  }
+}