Use case class ParsedDeltaSharingTablePath as return type for parsePa…

…th (#388)

* Update Python connector version to 0.8.0

* Setting version to 0.8.0

* Setting version to 0.8.1-SNAPSHOT

* Move parsePath from RemoteDeltaLog to DeltaSharingRestClient

* revert

* revert

* use case class instead of tuple as function return type

client/src/main/scala/io/delta/sharing/client/DeltaSharingClient.scala

@@ -75,6 +75,12 @@ trait DeltaSharingClient {
   def getProfileProvider: DeltaSharingProfileProvider = null
 }
 
+case class ParsedDeltaSharingTablePath(
+    profileFile: String,
+    share: String,
+    schema: String,
+    table: String)
+
 private[sharing] trait PaginationResponse {
   def nextPageToken: Option[String]
 }
@@ -895,7 +901,7 @@ object DeltaSharingRestClient extends Logging {
    * Parse the user provided path `profile_file#share.schema.share` to
    * `(profile_file, share, schema, share)`.
    */
-  def parsePath(path: String): (String, String, String, String) = {
+  def parsePath(path: String): ParsedDeltaSharingTablePath = {
     val shapeIndex = path.lastIndexOf('#')
     if (shapeIndex < 0) {
       throw new IllegalArgumentException(s"path $path is not valid")
@@ -909,7 +915,12 @@ object DeltaSharingRestClient extends Logging {
       tableSplits(1).isEmpty || tableSplits(2).isEmpty) {
       throw new IllegalArgumentException(s"path $path is not valid")
     }
-    (profileFile, tableSplits(0), tableSplits(1), tableSplits(2))
+    ParsedDeltaSharingTablePath(
+      profileFile = profileFile,
+      share = tableSplits(0),
+      schema = tableSplits(1),
+      table = tableSplits(2)
+    )
   }
 
   def apply(

client/src/main/scala/io/delta/sharing/client/DeltaSharingFileSystem.scala

@@ -65,6 +65,7 @@ private[sharing] class DeltaSharingFileSystem extends FileSystem {
 
   override def getUri(): URI = URI.create(s"$SCHEME:///")
 
+  // delta-sharing:///
   override def open(f: Path, bufferSize: Int): FSDataInputStream = {
     val path = DeltaSharingFileSystem.decode(f)
     val fetcher =

client/src/test/scala/io/delta/sharing/client/DeltaSharingRestClientSuite.scala

@@ -40,11 +40,12 @@ class DeltaSharingRestClientSuite extends DeltaSharingIntegrationTest {
   import DeltaSharingRestClient._
 
   test("parsePath") {
-    assert(DeltaSharingRestClient.parsePath("file:///foo/bar#a.b.c") == ("file:///foo/bar", "a", "b", "c"))
+    assert(DeltaSharingRestClient.parsePath("file:///foo/bar#a.b.c") ==
+      ParsedDeltaSharingTablePath("file:///foo/bar", "a", "b", "c"))
     assert(DeltaSharingRestClient.parsePath("file:///foo/bar#bar#a.b.c") ==
-      ("file:///foo/bar#bar", "a", "b", "c"))
+      ParsedDeltaSharingTablePath("file:///foo/bar#bar", "a", "b", "c"))
     assert(DeltaSharingRestClient.parsePath("file:///foo/bar#bar#a.b.c ") ==
-      ("file:///foo/bar#bar", "a", "b", "c "))
+      ParsedDeltaSharingTablePath("file:///foo/bar#bar", "a", "b", "c "))
     intercept[IllegalArgumentException] {
       DeltaSharingRestClient.parsePath("file:///foo/bar")
     }

spark/src/main/scala/io/delta/sharing/spark/RemoteDeltaLog.scala

@@ -105,9 +105,13 @@ private[sharing] object RemoteDeltaLog {
       path: String,
       forStreaming: Boolean = false,
       responseFormat: String = DeltaSharingOptions.RESPONSE_FORMAT_PARQUET): RemoteDeltaLog = {
-    val (profileFile, share, schema, table) = DeltaSharingRestClient.parsePath(path)
-    val client = DeltaSharingRestClient(profileFile, forStreaming, responseFormat)
-    val deltaSharingTable = DeltaSharingTable(name = table, schema = schema, share = share)
+    val parsedPath = DeltaSharingRestClient.parsePath(path)
+    val client = DeltaSharingRestClient(parsedPath.profileFile, forStreaming, responseFormat)
+    val deltaSharingTable = DeltaSharingTable(
+      name = parsedPath.table,
+      schema = parsedPath.schema,
+      share = parsedPath.share
+    )
     new RemoteDeltaLog(deltaSharingTable, new Path(path), client)
   }
 }