Refactor Spark Connector into two modules for reusability

pinot-connectors/pinot-spark-2-connector/pom.xml

@@ -0,0 +1,247 @@
+<?xml version="1.0"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <artifactId>pinot-connectors</artifactId>
+    <groupId>org.apache.pinot</groupId>
+    <version>0.13.0-SNAPSHOT</version>
+    <relativePath>..</relativePath>
+  </parent>
+  <artifactId>pinot-spark-2-connector</artifactId>
+  <name>Pinot Spark 2 Connector</name>
+  <url>https://pinot.apache.org/</url>
+  <properties>
+    <pinot.root>${basedir}/../..</pinot.root>
+    <spark.version>2.4.5</spark.version>
+    <paranamer.version>2.8</paranamer.version>
+    <scalaxml.version>1.3.0</scalaxml.version>
+    <scalatest.version>3.1.1</scalatest.version>
+
+    <!-- TODO: delete this prop once all the checkstyle warnings are fixed -->
+    <checkstyle.fail.on.violation>false</checkstyle.fail.on.violation>
+  </properties>
+
+  <profiles>
+    <profile>
+      <id>scala-2.12</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+      </activation>
+      <properties>
+        <scala.version>2.12.11</scala.version>
+        <scala.compat.version>2.12</scala.compat.version>
+      </properties>
+      <dependencies>
+        <dependency>
+          <groupId>org.scala-lang.modules</groupId>
+          <artifactId>scala-xml_${scala.compat.version}</artifactId>
+          <version>${scalaxml.version}</version>
+          <exclusions>
+            <exclusion>
+              <groupId>org.scala-lang</groupId>
+              <artifactId>scala-library</artifactId>
+            </exclusion>
+          </exclusions>
+        </dependency>
+        <dependency>
+          <groupId>org.apache.spark</groupId>
+          <artifactId>spark-sql_${scala.compat.version}</artifactId>
+          <version>${spark.version}</version>
+          <scope>provided</scope>
+          <exclusions>
+            <exclusion>
+              <groupId>org.apache.curator</groupId>
+              <artifactId>curator-recipes</artifactId>
+            </exclusion>
+            <exclusion>
+              <groupId>com.thoughtworks.paranamer</groupId>
+              <artifactId>paranamer</artifactId>
+            </exclusion>
+            <exclusion>
+              <groupId>org.scala-lang.modules</groupId>
+              <artifactId>scala-xml_${scala.compat.version}</artifactId>
+            </exclusion>
+            <exclusion>
+              <groupId>org.scala-lang</groupId>
+              <artifactId>scala-library</artifactId>
+            </exclusion>
+            <exclusion>
+              <groupId>com.zaxxer</groupId>
+              <artifactId>HikariCP-java7</artifactId>
+            </exclusion>
+          </exclusions>
+        </dependency>
+        <dependency>
+          <groupId>org.scala-lang</groupId>
+          <artifactId>scala-library</artifactId>
+          <version>${scala.version}</version>
+          <scope>provided</scope>
+        </dependency>
+        <!-- tests -->
+        <dependency>
+          <groupId>org.scalatest</groupId>
+          <artifactId>scalatest_${scala.compat.version}</artifactId>
+          <version>${scalatest.version}</version>
+          <scope>test</scope>
+          <exclusions>
+            <exclusion>
+              <groupId>org.scala-lang.modules</groupId>
+              <artifactId>scala-xml_${scala.compat.version}</artifactId>
+            </exclusion>
+            <exclusion>
+              <groupId>org.scala-lang</groupId>
+              <artifactId>scala-library</artifactId>
+            </exclusion>
+          </exclusions>
+        </dependency>
+      </dependencies>
+      <build>
+        <plugins>
+          <!-- scala build -->
+          <plugin>
+            <groupId>net.alchim31.maven</groupId>
+            <artifactId>scala-maven-plugin</artifactId>
+            <version>3.2.2</version>
+            <executions>
+              <execution>
+                <id>eclipse-add-source</id>
+                <goals>
+                  <goal>add-source</goal>
+                </goals>
+              </execution>
+              <execution>
+                <id>scala-compile-first</id>
+                <phase>process-resources</phase>
+                <goals>
+                  <goal>compile</goal>
+                </goals>
+              </execution>
+              <execution>
+                <id>scala-test-compile-first</id>
+                <phase>process-test-resources</phase>
+                <goals>
+                  <goal>testCompile</goal>
+                </goals>
+              </execution>
+              <execution>
+                <id>attach-scaladocs</id>
+                <phase>verify</phase>
+                <goals>
+                  <goal>doc-jar</goal>
+                </goals>
+              </execution>
+            </executions>
+            <configuration>
+              <scalaVersion>${scala.version}</scalaVersion>
+              <args>
+                <arg>-unchecked</arg>
+                <arg>-deprecation</arg>
+                <arg>-feature</arg>
+              </args>
+              <jvmArgs>
+                <jvmArg>-Xms1024m</jvmArg>
+                <jvmArg>-Xmx1024m</jvmArg>
+              </jvmArgs>
+              <javacArgs>
+                <javacArg>-source</javacArg>
+                <javacArg>${jdk.version}</javacArg>
+                <javacArg>-target</javacArg>
+                <javacArg>${jdk.version}</javacArg>
+                <javacArg>-Xlint:all,-serial,-path</javacArg>
+              </javacArgs>
+            </configuration>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+    <profile>
+      <id>release-sign-artifacts</id>
+      <build>
+        <plugins>
+          <plugin>
+            <!-- GPG signing. It's due to the fact that maven-source-plugin and scala-maven-plugin
+             have separate lifecycles, and the gpg files created by the former get deleted by the latter.
+             Thus, explicitly adding this plugin to a new profile to sign the files at the end all at once. -->
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-gpg-plugin</artifactId>
+            <version>1.6</version>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+
+  <build>
+    <sourceDirectory>src/main/scala</sourceDirectory>
+    <testSourceDirectory>src/test/scala</testSourceDirectory>
+    <resources>
+      <resource>
+        <directory>src/main/resources</directory>
+      </resource>
+    </resources>
+    <testResources>
+      <testResource>
+        <directory>src/test/resources</directory>
+      </testResource>
+    </testResources>
+    <plugins>
+      <!-- disable surefire -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <version>2.7</version>
+        <configuration>
+          <skipTests>true</skipTests>
+        </configuration>
+      </plugin>
+      <!-- enable scalatest -->
+      <plugin>
+        <groupId>org.scalatest</groupId>
+        <artifactId>scalatest-maven-plugin</artifactId>
+        <version>1.0</version>
+        <configuration>
+          <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+          <junitxml>.</junitxml>
+          <parallel>false</parallel>
+        </configuration>
+        <executions>
+          <execution>
+            <id>test</id>
+            <goals>
+              <goal>test</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.pinot</groupId>
+      <artifactId>pinot-spark-common</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+  </dependencies>
+</project>

pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceReader.scala → pinot-connectors/pinot-spark-2-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceReader.scala

@@ -20,8 +20,10 @@ package org.apache.pinot.connector.spark.datasource
 
 import java.util.{List => JList}
 
-import org.apache.pinot.connector.spark.connector._
-import org.apache.pinot.connector.spark.connector.query.SQLSelectionQueryGenerator
+import org.apache.pinot.connector.spark.common.{InstanceInfo, PinotClusterClient, PinotDataSourceReadOptions}
+import org.apache.pinot.connector.spark.common.query.ScanQueryGenerator
+import org.apache.pinot.connector.spark.common.partition.PinotSplitter
+import org.apache.pinot.connector.spark.datasource.query.FilterPushDown
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.sources._
 import org.apache.spark.sql.sources.v2.DataSourceOptions
@@ -44,7 +46,7 @@ class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[Struc
   with SupportsPushDownFilters
   with SupportsPushDownRequiredColumns {
 
-  private val readParameters = PinotDataSourceReadOptions.from(options)
+  private val readParameters = PinotDataSourceReadOptions.from(options.asMap())
   private var acceptedFilters: Array[Filter] = Array.empty
   private var currentSchema: StructType = _
 
@@ -53,7 +55,7 @@ class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[Struc
       currentSchema = userSchema.getOrElse {
         val pinotTableSchema =
           PinotClusterClient.getTableSchema(readParameters.controller, readParameters.tableName)
-        PinotUtils.pinotSchemaToSparkSchema(pinotTableSchema)
+        TypeConverter.pinotSchemaToSparkSchema(pinotTableSchema)
       }
     }
     currentSchema
@@ -71,7 +73,7 @@ class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[Struc
       }
 
     val whereCondition = FilterPushDown.compileFiltersToSqlWhereClause(this.acceptedFilters)
-    val generatedSQLs = SQLSelectionQueryGenerator.generate(
+    val generatedSQLs = ScanQueryGenerator.generate(
       readParameters.tableName,
       readParameters.tableType,
       timeBoundaryInfo,

pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotInputPartition.scala → pinot-connectors/pinot-spark-2-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotInputPartition.scala

@@ -18,8 +18,10 @@
  */
 package org.apache.pinot.connector.spark.datasource
 
-import org.apache.pinot.connector.spark.connector.PinotSplit
-import org.apache.spark.internal.Logging
+import org.apache.pinot.common.datatable.DataTable
+import org.apache.pinot.connector.spark.common.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.common.partition.PinotSplit
+import org.apache.pinot.connector.spark.common.reader.PinotAbstractPartitionReader
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.sources.v2.reader.{InputPartition, InputPartitionReader}
 import org.apache.spark.sql.types.StructType
@@ -29,10 +31,14 @@ class PinotInputPartition(
     partitionId: Int,
     pinotSplit: PinotSplit,
     dataSourceOptions: PinotDataSourceReadOptions)
-  extends InputPartition[InternalRow]
-  with Logging {
+  extends InputPartition[InternalRow] {
 
   override def createPartitionReader(): InputPartitionReader[InternalRow] = {
-    new PinotInputPartitionReader(schema, partitionId, pinotSplit, dataSourceOptions)
+    new InputPartitionReader[InternalRow] with PinotAbstractPartitionReader[InternalRow]  {
+      override def _partitionId: Int = partitionId
+      override def _pinotSplit: PinotSplit = pinotSplit
+      override def _dataSourceOptions: PinotDataSourceReadOptions = dataSourceOptions
+      override def _translator: DataTable => Seq[InternalRow] = TypeConverter.pinotDataTableToInternalRows(_, schema)
+    }
   }
 }

pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotUtils.scala → pinot-connectors/pinot-spark-2-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/TypeConverter.scala

@@ -16,11 +16,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.connector.spark.connector
+package org.apache.pinot.connector.spark.datasource
 
 import org.apache.pinot.common.datatable.DataTable
 import org.apache.pinot.common.utils.DataSchema.ColumnDataType
-import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.common.PinotException
 import org.apache.pinot.spi.data.{FieldSpec, Schema}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.util.ArrayData
@@ -32,7 +32,7 @@ import scala.collection.JavaConverters._
 /**
  * Helper methods for spark-pinot conversions
  */
-private[pinot] object PinotUtils {
+private[datasource] object TypeConverter {
 
   /** Convert a Pinot schema to Spark schema. */
   def pinotSchemaToSparkSchema(schema: Schema): StructType = {

pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/FilterPushDown.scala → pinot-connectors/pinot-spark-2-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/query/FilterPushDown.scala

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.connector.spark.connector
+package org.apache.pinot.connector.spark.datasource.query
 
 import java.sql.{Date, Timestamp}
 

pinot-connectors/pinot-spark-2-connector/src/test/scala/org/apache/pinot/connector/spark/datasource/BaseTest.scala

@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+
+trait BaseTest extends AnyFunSuite with Matchers

pinot-connectors/pinot-spark-connector/src/test/scala/org/apache/pinot/connector/spark/ExampleSparkPinotConnectorTest.scala → pinot-connectors/pinot-spark-2-connector/src/test/scala/org/apache/pinot/connector/spark/datasource/ExampleSparkPinotConnectorTest.scala

@@ -16,9 +16,9 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pinot.connector.spark
+package org.apache.pinot.connector.spark.datasource
 
-import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.connector.spark.common.Logging
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.types.{DataTypes, StructField, StructType}