save so far

Author: brkyvz

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/TableCapabilityCheck.scala

@@ -20,13 +20,9 @@ package org.apache.spark.sql.execution.datasources.v2
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.expressions.Literal
 import org.apache.spark.sql.catalyst.plans.logical.{AppendData, LogicalPlan, OverwriteByExpression, OverwritePartitionsDynamic}
+import org.apache.spark.sql.connector.catalog.{SupportsWrite, Table}
 import org.apache.spark.sql.connector.catalog.TableCapability._
 import org.apache.spark.sql.execution.streaming.{StreamingRelation, StreamingRelationV2}
-<<<<<<< HEAD
-import org.apache.spark.sql.sources.v2.{SupportsWrite, Table}
-import org.apache.spark.sql.sources.v2.TableCapability._
-=======
->>>>>>> c56a012bc839cd2f92c2be41faea91d1acfba4eb
 import org.apache.spark.sql.types.BooleanType
 
 /**
@@ -37,9 +33,8 @@ object TableCapabilityCheck extends (LogicalPlan => Unit) {
 
   private def failAnalysis(msg: String): Unit = throw new AnalysisException(msg)
 
-  private def supportsBatchWrite(table: Table): Boolean = table match {
-    case supportsWrite: SupportsWrite => supportsWrite.supportsAny(BATCH_WRITE, V1_BATCH_WRITE)
-    case _ => false
+  private def supportsBatchWrite(table: Table): Boolean = {
+    table.supportsAny(BATCH_WRITE, V1_BATCH_WRITE)
   }
 
   override def apply(plan: LogicalPlan): Unit = {

sql/core/src/test/scala/org/apache/spark/sql/connector/V1WriteFallbackSuite.scala

@@ -28,6 +28,7 @@ import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row, SQLCo
 import org.apache.spark.sql.connector.catalog.{SupportsWrite, Table, TableCapability, TableProvider}
 import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTransform, Transform}
 import org.apache.spark.sql.connector.write.{SupportsOverwrite, SupportsTruncate, V1WriteBuilder, WriteBuilder}
+import org.apache.spark.sql.execution.datasources.{DataSource, DataSourceUtils}
 import org.apache.spark.sql.sources._
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
@@ -114,8 +115,12 @@ private object InMemoryV1Provider {
   }
 }
 
-class InMemoryV1Provider extends TableProvider with DataSourceRegister {
+class InMemoryV1Provider
+  extends TableProvider
+  with DataSourceRegister
+  with CreatableRelationProvider {
   override def getTable(options: CaseInsensitiveStringMap): Table = {
+
     InMemoryV1Provider.tables.getOrElse(options.get("name"), {
       new InMemoryTableWithV1Fallback(
         "InMemoryTableWithV1Fallback",
@@ -127,6 +132,45 @@ class InMemoryV1Provider extends TableProvider with DataSourceRegister {
   }
 
   override def shortName(): String = "in-memory"
+
+  override def createRelation(
+      sqlContext: SQLContext,
+      mode: SaveMode,
+      parameters: Map[String, String],
+      data: DataFrame): BaseRelation = {
+    val _sqlContext = sqlContext
+
+    val partitioning = parameters.get(DataSourceUtils.PARTITIONING_COLUMNS_KEY).map { value =>
+      DataSourceUtils.decodePartitioningColumns(value).map { partitioningColumn =>
+
+      }
+    }
+
+    val table = new InMemoryTableWithV1Fallback(
+      "InMemoryTableWithV1Fallback",
+      data.schema.asNullable,
+      Array.empty,
+      Map.empty[String, String].asJava
+    )
+
+    def getRelation: BaseRelation = new BaseRelation {
+      override def sqlContext: SQLContext = _sqlContext
+      override def schema: StructType = table.schema
+    }
+
+    if (mode == SaveMode.ErrorIfExists && dataMap.nonEmpty) {
+      throw new AnalysisException("Table already exists")
+    } else if (mode == SaveMode.Ignore && dataMap.nonEmpty) {
+      // do nothing
+      return getRelation
+    }
+    val writer = new FallbackWriteBuilder(new CaseInsensitiveStringMap(parameters.asJava))
+    if (mode == SaveMode.Overwrite) {
+      writer.truncate()
+    }
+    writer.buildForV1Write().insert(data, overwrite = false)
+    getRelation
+  }
 }
 
 class InMemoryTableWithV1Fallback(
@@ -135,8 +179,7 @@ class InMemoryTableWithV1Fallback(
     override val partitioning: Array[Transform],
     override val properties: util.Map[String, String])
   extends Table
-  with SupportsWrite
-  with CreatableRelationProvider {
+  with SupportsWrite {
 
   partitioning.foreach { t =>
     if (!t.isInstanceOf[IdentityTransform]) {
@@ -159,24 +202,6 @@ class InMemoryTableWithV1Fallback(
     new FallbackWriteBuilder(options)
   }
 
-  override def createRelation(
-      sqlContext: SQLContext,
-      mode: SaveMode,
-      parameters: Map[String, String],
-      data: DataFrame): BaseRelation = {
-    if (mode == SaveMode.ErrorIfExists && dataMap.nonEmpty) {
-      throw new AnalysisException("Table already exists")
-    } else if (mode == SaveMode.Ignore && dataMap.nonEmpty) {
-      // do nothing
-    } else if (mode == SaveMode.Overwrite) {
-      val writer = new FallbackWriteBuilder(new CaseInsensitiveStringMap(parameters.asJava))
-      writer.truncate()
-      writer.buildForV1Write().insert(data, overwrite = false)
-    } else {
-
-    }
-  }
-
   private class FallbackWriteBuilder(options: CaseInsensitiveStringMap)
     extends WriteBuilder
     with V1WriteBuilder