Skip to content

[SPARK-34498][SQL][TESTS] fix the remaining problems in #31560 #31621

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kevincmchen wants to merge 3 commits into from
Closed

[SPARK-34498][SQL][TESTS] fix the remaining problems in #31560 #31621

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e252d75
fix the remaining problems in SPARK-34432.
kevincmchen Feb 23, 2021
7f82afc
extend TestingV2Source instead of SimpleTableProvider
kevincmchen Mar 1, 2021
6c38617
it's unnecessary to override readSchema in MyScanBuilder and schema i…
kevincmchen Mar 2, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 1 addition & 7 deletions sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleWritableDataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
import org.apache.spark.sql.connector.SimpleCounter;
import org.apache.spark.sql.connector.TestingV2Source;
import org.apache.spark.sql.connector.catalog.SessionConfigSupport;
import org.apache.spark.sql.connector.catalog.SupportsWrite;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCapability;
Expand All @@ -50,12 +49,7 @@
* Each task writes data to `target/_temporary/uniqueId/$jobId-$partitionId-$attemptNumber`.
* Each job moves files from `target/_temporary/uniqueId/` to `target`.
*/
public class JavaSimpleWritableDataSource implements TestingV2Source, SessionConfigSupport {

@Override
public String keyPrefix() {
return "javaSimpleWritableDataSource";
}
public class JavaSimpleWritableDataSource implements TestingV2Source {

static class MyScanBuilder extends JavaSimpleScanBuilder {

Expand Down
17 changes: 6 additions & 11 deletions sql/core/src/test/scala/org/apache/spark/sql/connector/SimpleWritableDataSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,10 @@ import org.apache.hadoop.fs.{FileSystem, Path}

import org.apache.spark.SparkContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.catalog.{SessionConfigSupport, SupportsWrite, Table, TableCapability}
import org.apache.spark.sql.connector.catalog.{SupportsWrite, Table, TableCapability}
import org.apache.spark.sql.connector.catalog.TableCapability._
import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader, PartitionReaderFactory, ScanBuilder}
import org.apache.spark.sql.connector.write._
import org.apache.spark.sql.internal.connector.SimpleTableProvider
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.SerializableConfiguration
Expand All @@ -41,11 +40,7 @@ import org.apache.spark.util.SerializableConfiguration
* Each task writes data to `target/_temporary/uniqueId/$jobId-$partitionId-$attemptNumber`.
* Each job moves files from `target/_temporary/uniqueId/` to `target`.
*/
class SimpleWritableDataSource extends SimpleTableProvider with SessionConfigSupport {

private val tableSchema = new StructType().add("i", "long").add("j", "long")

override def keyPrefix: String = "simpleWritableDataSource"
class SimpleWritableDataSource extends TestingV2Source {

class MyScanBuilder(path: String, conf: Configuration) extends SimpleScanBuilder {
override def planInputPartitions(): Array[InputPartition] = {
Expand All @@ -68,7 +63,7 @@ class SimpleWritableDataSource extends SimpleTableProvider with SessionConfigSup
new CSVReaderFactory(serializableConf)
}

override def readSchema(): StructType = tableSchema
override def readSchema(): StructType = TestingV2Source.schema
Copy link
Contributor

@cloud-fan cloud-fan Mar 1, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SimpleScanBuilder.readSchema is already TestingV2Source.schema, we don't need to override it here.

Copy link
Contributor Author

@kevincmchen kevincmchen Mar 2, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, it‘s unnecessary to override the function.

}

class MyWriteBuilder(path: String, info: LogicalWriteInfo)
Expand Down Expand Up @@ -134,7 +129,7 @@ class SimpleWritableDataSource extends SimpleTableProvider with SessionConfigSup
private val path = options.get("path")
private val conf = SparkContext.getActive.get.hadoopConfiguration

override def schema(): StructType = tableSchema
override def schema(): StructType = TestingV2Source.schema
Copy link
Contributor

@cloud-fan cloud-fan Mar 1, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto, SimpleBatchTable.schema is already TestingV2Source.schema

Copy link
Contributor Author

@kevincmchen kevincmchen Mar 2, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, it‘s unnecessary.


override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
new MyScanBuilder(new Path(path).toUri.toString, conf)
Expand Down Expand Up @@ -179,7 +174,7 @@ class CSVReaderFactory(conf: SerializableConfiguration)
}
}

override def get(): InternalRow = InternalRow(currentLine.split(",").map(_.trim.toLong): _*)
override def get(): InternalRow = InternalRow(currentLine.split(",").map(_.trim.toInt): _*)

override def close(): Unit = {
inputStream.close()
Expand Down Expand Up @@ -222,7 +217,7 @@ class CSVDataWriter(fs: FileSystem, file: Path) extends DataWriter[InternalRow]
private val out = fs.create(file)

override def write(record: InternalRow): Unit = {
out.writeBytes(s"${record.getLong(0)},${record.getLong(1)}\n")
out.writeBytes(s"${record.getInt(0)},${record.getInt(1)}\n")
Copy link
Member

@srowen srowen Mar 1, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change just matches the TestingV2Source.schema?

Copy link
Contributor

@cloud-fan cloud-fan Mar 1, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yea

}

override def commit(): WriterCommitMessage = {
Expand Down