[Spark][Sharing] Fix Delta Sharing DataFrame not updated for Snapshot Query

Cherrypick #2574 to branch-3.1

Fix Delta Sharing DataFrame not updated for Snapshot Query:
When provider updated the table(insert or delete row), in delta-sharing-spark session, the dataframe on the same query for the same table is not updated, though a new rpc is made, the local delta log is updated, but it’s still at version 0.

Author: linzhou-db

build.sbt

@@ -224,7 +224,7 @@ lazy val sharing = (project in file("sharing"))
     libraryDependencies ++= Seq(
       "org.apache.spark" %% "spark-sql" % sparkVersion % "provided",
 
-      "io.delta" %% "delta-sharing-client" % "1.0.3",
+      "io.delta" %% "delta-sharing-client" % "1.0.4",
 
       // Test deps
       "org.scalatest" %% "scalatest" % scalaTestVersion % "test",

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

@@ -391,7 +391,7 @@ private[sharing] class DeltaSharingDataSource
     val params = new DeltaSharingFileIndexParams(
       new Path(path),
       spark,
-      deltaSharingTableMetadata.metadata,
+      deltaSharingTableMetadata,
       options
     )
     if (ConfUtils.limitPushdownEnabled(spark.sessionState.conf)) {

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

@@ -36,7 +36,7 @@ import org.apache.spark.sql.types.StructType
 private[sharing] case class DeltaSharingFileIndexParams(
     path: Path,
     spark: SparkSession,
-    metadata: model.DeltaSharingMetadata,
+    deltaSharingTableMetadata: DeltaSharingUtils.DeltaSharingTableMetadata,
     options: DeltaSharingOptions)
 
 /**
@@ -62,18 +62,20 @@ case class DeltaSharingFileIndex(
   override def refresh(): Unit = {}
 
   override def sizeInBytes: Long =
-    Option(params.metadata.size).getOrElse {
+    Option(params.deltaSharingTableMetadata.metadata.size).getOrElse {
       // Throw error if metadata.size is not returned, to urge the server to respond a table size.
       throw new IllegalStateException(
         "size is null in the metadata returned from the delta " +
-        s"sharing server: ${params.metadata}."
+        s"sharing server: ${params.deltaSharingTableMetadata.metadata}."
       )
     }
 
-  override def partitionSchema: StructType = params.metadata.partitionSchema
+  override def partitionSchema: StructType =
+    params.deltaSharingTableMetadata.metadata.partitionSchema
 
   // Returns the partition columns of the shared delta table based on the returned metadata.
-  def partitionColumns: Seq[String] = params.metadata.deltaMetadata.partitionColumns
+  def partitionColumns: Seq[String] =
+    params.deltaSharingTableMetadata.metadata.deltaMetadata.partitionColumns
 
   override def rootPaths: Seq[Path] = params.path :: Nil
 
@@ -98,7 +100,8 @@ case class DeltaSharingFileIndex(
       // keeps the string the same for the same filters.
       partitionFilters.map(_.sql).mkString(";"),
       dataFilters.map(_.sql).mkString(";"),
-      jsonPredicateHints.getOrElse("")
+      jsonPredicateHints.getOrElse(""),
+      params.deltaSharingTableMetadata.version
     )
     val tablePathWithHashIdSuffix = DeltaSharingUtils.getTablePathWithIdSuffix(
       queryCustomTablePath,

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

@@ -276,9 +276,10 @@ object DeltaSharingUtils extends Logging {
       options: DeltaSharingOptions,
       partitionFiltersString: String,
       dataFiltersString: String,
-      jsonPredicateHints: String): String = {
+      jsonPredicateHints: String,
+      version: Long): String = {
     val fullQueryString = s"${options.versionAsOf}_${options.timestampAsOf}_" +
-      s"${partitionFiltersString}_${dataFiltersString}_${jsonPredicateHints}"
+      s"${partitionFiltersString}_${dataFiltersString}_${jsonPredicateHints}_${version}"
     Hashing.sha256().hashString(fullQueryString, UTF_8).toString
   }
 

sharing/src/test/scala/io/delta/sharing/spark/DeltaSharingDataSourceDeltaSuite.scala

@@ -199,6 +199,87 @@ trait DeltaSharingDataSourceDeltaSuiteBase
     }
   }
 
+  test("DeltaSharingDataSource able to read data with changes") {
+    withTempDir { tempDir =>
+      val deltaTableName = "delta_table_change"
+
+      def test(tablePath: String, expectedCount: Int, expectedSchema: StructType): Unit = {
+        assert(
+          expectedSchema == spark.read
+            .format("deltaSharing")
+            .option("responseFormat", "delta")
+            .load(tablePath)
+            .schema
+        )
+
+        val deltaDf = spark.read.format("delta").table(deltaTableName)
+        val sharingDf =
+          spark.read.format("deltaSharing").option("responseFormat", "delta").load(tablePath)
+        checkAnswer(deltaDf, sharingDf)
+        assert(sharingDf.count() == expectedCount)
+      }
+
+      withTable(deltaTableName) {
+        val sharedTableName = "shared_table_change"
+        createTable(deltaTableName)
+
+        // test 1: insert 2 rows
+        sql(
+          s"INSERT INTO $deltaTableName" +
+            """ VALUES (1, "one", "2023-01-01", "2023-01-01 00:00:00"),
+              |(2, "two", "2023-02-02", "2023-02-02 00:00:00")""".stripMargin
+        )
+        prepareMockedClientAndFileSystemResult(deltaTableName, sharedTableName)
+        prepareMockedClientGetTableVersion(deltaTableName, sharedTableName)
+        val expectedSchema: StructType = new StructType()
+          .add("c1", IntegerType)
+          .add("c2", StringType)
+          .add("c3", DateType)
+          .add("c4", TimestampType)
+        withSQLConf(getDeltaSharingClassesSQLConf.toSeq: _*) {
+          val profileFile = prepareProfileFile(tempDir)
+          val tableName = s"share1.default.$sharedTableName"
+          test(s"${profileFile.getCanonicalPath}#$tableName", 2, expectedSchema)
+        }
+
+        // test 2: insert 2 more rows, and rename a column
+        spark.sql(
+          s"""ALTER TABLE $deltaTableName SET TBLPROPERTIES('delta.minReaderVersion' = '2',
+             |'delta.minWriterVersion' = '5',
+             |'delta.columnMapping.mode' = 'name', 'delta.enableDeletionVectors' = true)""".stripMargin
+        )
+        sql(
+          s"INSERT INTO $deltaTableName" +
+            """ VALUES (3, "three", "2023-03-03", "2023-03-03 00:00:00"),
+              |(4, "four", "2023-04-04", "2023-04-04 00:00:00")""".stripMargin
+        )
+        sql(s"""ALTER TABLE $deltaTableName RENAME COLUMN c3 TO c3rename""")
+        prepareMockedClientAndFileSystemResult(deltaTableName, sharedTableName)
+        prepareMockedClientGetTableVersion(deltaTableName, sharedTableName)
+        val expectedNewSchema: StructType = new StructType()
+          .add("c1", IntegerType)
+          .add("c2", StringType)
+          .add("c3rename", DateType)
+          .add("c4", TimestampType)
+        withSQLConf(getDeltaSharingClassesSQLConf.toSeq: _*) {
+          val profileFile = prepareProfileFile(tempDir)
+          val tableName = s"share1.default.$sharedTableName"
+          test(s"${profileFile.getCanonicalPath}#$tableName", 4, expectedNewSchema)
+        }
+
+        // test 3: delete 1 row
+        sql(s"DELETE FROM $deltaTableName WHERE c1 = 2")
+        prepareMockedClientAndFileSystemResult(deltaTableName, sharedTableName)
+        prepareMockedClientGetTableVersion(deltaTableName, sharedTableName)
+        withSQLConf(getDeltaSharingClassesSQLConf.toSeq: _*) {
+          val profileFile = prepareProfileFile(tempDir)
+          val tableName = s"share1.default.$sharedTableName"
+          test(s"${profileFile.getCanonicalPath}#$tableName", 3, expectedNewSchema)
+        }
+      }
+    }
+  }
+
   test("DeltaSharingDataSource able to auto resolve responseFormat") {
     withTempDir { tempDir =>
       val deltaTableName = "delta_table_auto"

sharing/src/test/scala/io/delta/sharing/spark/DeltaSharingFileIndexSuite.scala

@@ -59,6 +59,8 @@ private object TestUtils {
   }
 
   // scalastyle:off line.size.limit
+  val protocolStr =
+    """{"protocol":{"deltaProtocol":{"minReaderVersion": 1, "minWriterVersion": 1}}}"""
   val metaDataStr =
     """{"metaData":{"size":809,"deltaMetadata":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"c1\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"c2\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["c2"],"configuration":{},"createdTime":1691734718560}}}"""
   val metaDataWithoutSizeStr =
@@ -148,7 +150,7 @@ class TestDeltaSharingClientForFileIndex(
     DeltaTableFiles(
       version = 0,
       lines = Seq[String](
-        """{"protocol":{"deltaProtocol":{"minReaderVersion": 1, "minWriterVersion": 1}}}""",
+        protocolStr,
         metaDataStr,
         getAddFileStr1(paths(numGetFileCalls.min(1)), urlExpirationMsOpt),
         getAddFileStr2(urlExpirationMsOpt)
@@ -220,7 +222,11 @@ class DeltaSharingFileIndexSuite
     val params = new DeltaSharingFileIndexParams(
       tablePath,
       spark,
-      getMockedDeltaSharingMetadata(metaData),
+      DeltaSharingUtils.DeltaSharingTableMetadata(
+        version = 0,
+        protocol = JsonUtils.fromJson[model.DeltaSharingSingleAction](protocolStr).protocol,
+        metadata = getMockedDeltaSharingMetadata(metaData)
+      ),
       new DeltaSharingOptions(Map("path" -> tablePath.toString))
     )
     val dsTable = Table(share = shareName, schema = schemaName, name = sharedTableName)