[spark] Introduce paimon_incremental_between_timestamp and paimon_incremental_to_auto_tag tvf (apache#4855)

Author: Zouxxyy

docs/content/spark/sql-query.md

@@ -77,11 +77,17 @@ You can also force specifying `'incremental-between-scan-mode'`.
 
 Paimon supports that use Spark SQL to do the incremental query that implemented by Spark Table Valued Function.
 
-you can use `paimon_incremental_query` in query to extract the incremental data:
-
 ```sql
 -- read the incremental data between snapshot id 12 and snapshot id 20.
 SELECT * FROM paimon_incremental_query('tableName', 12, 20);
+
+-- read the incremental data between ts 1692169900000 and ts 1692169900000.
+SELECT * FROM paimon_incremental_between_timestamp('tableName', '1692169000000', '1692169900000');
+
+-- read the incremental data to tag '2024-12-04'.
+-- Paimon will find an earlier tag and return changes between them.
+-- If the tag doesn't exist or the earlier tag doesn't exist, return empty.
+SELECT * FROM paimon_incremental_to_auto_tag('tableName', '2024-12-04');
 ```
 
 In Batch SQL, the `DELETE` records are not allowed to be returned, so records of `-D` will be dropped.

docs/layouts/shortcodes/generated/core_configuration.html

@@ -396,25 +396,25 @@
             <td><h5>incremental-between</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>Read incremental changes between start snapshot (exclusive) and end snapshot, for example, '5,10' means changes between snapshot 5 and snapshot 10.</td>
+            <td>Read incremental changes between start snapshot (exclusive) and end snapshot (inclusive), for example, '5,10' means changes between snapshot 5 and snapshot 10.</td>
         </tr>
         <tr>
             <td><h5>incremental-between-scan-mode</h5></td>
             <td style="word-wrap: break-word;">auto</td>
             <td><p>Enum</p></td>
-            <td>Scan kind when Read incremental changes between start snapshot (exclusive) and end snapshot. <br /><br />Possible values:<ul><li>"auto": Scan changelog files for the table which produces changelog files. Otherwise, scan newly changed files.</li><li>"delta": Scan newly changed files between snapshots.</li><li>"changelog": Scan changelog files between snapshots.</li></ul></td>
+            <td>Scan kind when Read incremental changes between start snapshot (exclusive) and end snapshot (inclusive). <br /><br />Possible values:<ul><li>"auto": Scan changelog files for the table which produces changelog files. Otherwise, scan newly changed files.</li><li>"delta": Scan newly changed files between snapshots.</li><li>"changelog": Scan changelog files between snapshots.</li></ul></td>
         </tr>
         <tr>
             <td><h5>incremental-between-timestamp</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>Read incremental changes between start timestamp (exclusive) and end timestamp, for example, 't1,t2' means changes between timestamp t1 and timestamp t2.</td>
+            <td>Read incremental changes between start timestamp (exclusive) and end timestamp (inclusive), for example, 't1,t2' means changes between timestamp t1 and timestamp t2.</td>
         </tr>
         <tr>
             <td><h5>incremental-to-auto-tag</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>Used to specify the auto-created tag to reading incremental changes.</td>
+            <td>Used to specify the end tag (inclusive), and Paimon will find an earlier tag and return changes between them. If the tag doesn't exist or the earlier tag doesn't exist, return empty. </td>
         </tr>
         <tr>
             <td><h5>local-merge-buffer-size</h5></td>

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1089,30 +1089,31 @@ public class CoreOptions implements Serializable {
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "Read incremental changes between start snapshot (exclusive) and end snapshot, "
+                            "Read incremental changes between start snapshot (exclusive) and end snapshot (inclusive), "
                                     + "for example, '5,10' means changes between snapshot 5 and snapshot 10.");
 
     public static final ConfigOption<IncrementalBetweenScanMode> INCREMENTAL_BETWEEN_SCAN_MODE =
             key("incremental-between-scan-mode")
                     .enumType(IncrementalBetweenScanMode.class)
                     .defaultValue(IncrementalBetweenScanMode.AUTO)
                     .withDescription(
-                            "Scan kind when Read incremental changes between start snapshot (exclusive) and end snapshot. ");
+                            "Scan kind when Read incremental changes between start snapshot (exclusive) and end snapshot (inclusive). ");
 
     public static final ConfigOption<String> INCREMENTAL_BETWEEN_TIMESTAMP =
             key("incremental-between-timestamp")
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "Read incremental changes between start timestamp (exclusive) and end timestamp, "
+                            "Read incremental changes between start timestamp (exclusive) and end timestamp (inclusive), "
                                     + "for example, 't1,t2' means changes between timestamp t1 and timestamp t2.");
 
     public static final ConfigOption<String> INCREMENTAL_TO_AUTO_TAG =
             key("incremental-to-auto-tag")
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "Used to specify the auto-created tag to reading incremental changes.");
+                            "Used to specify the end tag (inclusive), and Paimon will find an earlier tag and return changes between them. "
+                                    + "If the tag doesn't exist or the earlier tag doesn't exist, return empty. ");
 
     public static final ConfigOption<Boolean> END_INPUT_CHECK_PARTITION_EXPIRE =
             key("end-input.check-partition-expire")

paimon-docs/README.md

@@ -28,7 +28,7 @@ The `@ConfigGroups` annotation can be used to generate multiple files from a sin
 
 To integrate an `*Options` class from another package, add another module-package argument pair to `ConfigOptionsDocGenerator#LOCATIONS`.
 
-The files can be generated by running `mvn clean install -DskipTests` and `mvn package -Pgenerate-docs -pl paimon-docs -nsu -DskipTests`, and can be integrated into the documentation using `{{ include generated/<file-name> >}}`.
+The files can be generated by running `mvn package -Pgenerate-docs -pl paimon-docs -nsu -DskipTests -am`, and can be integrated into the documentation using `{{ include generated/<file-name> >}}`.
 
 **NOTE:** You need to make sure that the changed jar has been installed in the local maven repository.
 

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala

@@ -19,6 +19,7 @@
 package org.apache.paimon.spark.catalyst.plans.logical
 
 import org.apache.paimon.CoreOptions
+import org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions._
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.FunctionIdentifier
@@ -35,15 +36,22 @@ import scala.collection.JavaConverters._
 object PaimonTableValuedFunctions {
 
   val INCREMENTAL_QUERY = "paimon_incremental_query"
+  val INCREMENTAL_BETWEEN_TIMESTAMP = "paimon_incremental_between_timestamp"
+  val INCREMENTAL_TO_AUTO_TAG = "paimon_incremental_to_auto_tag"
 
-  val supportedFnNames: Seq[String] = Seq(INCREMENTAL_QUERY)
+  val supportedFnNames: Seq[String] =
+    Seq(INCREMENTAL_QUERY, INCREMENTAL_BETWEEN_TIMESTAMP, INCREMENTAL_TO_AUTO_TAG)
 
   private type TableFunctionDescription = (FunctionIdentifier, ExpressionInfo, TableFunctionBuilder)
 
   def getTableValueFunctionInjection(fnName: String): TableFunctionDescription = {
     val (info, builder) = fnName match {
       case INCREMENTAL_QUERY =>
         FunctionRegistryBase.build[IncrementalQuery](fnName, since = None)
+      case INCREMENTAL_BETWEEN_TIMESTAMP =>
+        FunctionRegistryBase.build[IncrementalBetweenTimestamp](fnName, since = None)
+      case INCREMENTAL_TO_AUTO_TAG =>
+        FunctionRegistryBase.build[IncrementalToAutoTag](fnName, since = None)
       case _ =>
         throw new Exception(s"Function $fnName isn't a supported table valued function.")
     }
@@ -58,12 +66,23 @@ object PaimonTableValuedFunctions {
 
     val sessionState = spark.sessionState
     val catalogManager = sessionState.catalogManager
-    val sparkCatalog = catalogManager.currentCatalog.asInstanceOf[TableCatalog]
-    val tableId = sessionState.sqlParser.parseTableIdentifier(args.head.eval().toString)
-    val namespace = tableId.database.map(Array(_)).getOrElse(catalogManager.currentNamespace)
-    val ident = Identifier.of(namespace, tableId.table)
+
+    val identifier = args.head.eval().toString
+    val (catalogName, dbName, tableName) = {
+      sessionState.sqlParser.parseMultipartIdentifier(identifier) match {
+        case Seq(table) =>
+          (catalogManager.currentCatalog.name(), catalogManager.currentNamespace.head, table)
+        case Seq(db, table) => (catalogManager.currentCatalog.name(), db, table)
+        case Seq(catalog, db, table) => (catalog, db, table)
+        case _ => throw new RuntimeException(s"Invalid table identifier: $identifier")
+      }
+    }
+
+    val sparkCatalog = catalogManager.catalog(catalogName).asInstanceOf[TableCatalog]
+    val ident: Identifier = Identifier.of(Array(dbName), tableName)
     val sparkTable = sparkCatalog.loadTable(ident)
     val options = tvf.parseArgs(args.tail)
+
     DataSourceV2Relation.create(
       sparkTable,
       Some(sparkCatalog),
@@ -87,20 +106,46 @@ abstract class PaimonTableValueFunction(val fnName: String) extends LeafNode {
   val args: Seq[Expression]
 
   def parseArgs(args: Seq[Expression]): Map[String, String]
-
 }
 
-/** Plan for the "paimon_incremental_query" function */
+/** Plan for the [[INCREMENTAL_QUERY]] function */
 case class IncrementalQuery(override val args: Seq[Expression])
-  extends PaimonTableValueFunction(PaimonTableValuedFunctions.INCREMENTAL_QUERY) {
+  extends PaimonTableValueFunction(INCREMENTAL_QUERY) {
 
   override def parseArgs(args: Seq[Expression]): Map[String, String] = {
     assert(
-      args.size >= 1 && args.size <= 2,
-      "paimon_incremental_query needs two parameters: startSnapshotId, and endSnapshotId.")
+      args.size == 2,
+      s"$INCREMENTAL_QUERY needs two parameters: startSnapshotId, and endSnapshotId.")
 
     val start = args.head.eval().toString
     val end = args.last.eval().toString
     Map(CoreOptions.INCREMENTAL_BETWEEN.key -> s"$start,$end")
   }
 }
+
+/** Plan for the [[INCREMENTAL_BETWEEN_TIMESTAMP]] function */
+case class IncrementalBetweenTimestamp(override val args: Seq[Expression])
+  extends PaimonTableValueFunction(INCREMENTAL_BETWEEN_TIMESTAMP) {
+
+  override def parseArgs(args: Seq[Expression]): Map[String, String] = {
+    assert(
+      args.size == 2,
+      s"$INCREMENTAL_BETWEEN_TIMESTAMP needs two parameters: startTimestamp, and endTimestamp.")
+
+    val start = args.head.eval().toString
+    val end = args.last.eval().toString
+    Map(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key -> s"$start,$end")
+  }
+}
+
+/** Plan for the [[INCREMENTAL_TO_AUTO_TAG]] function */
+case class IncrementalToAutoTag(override val args: Seq[Expression])
+  extends PaimonTableValueFunction(INCREMENTAL_TO_AUTO_TAG) {
+
+  override def parseArgs(args: Seq[Expression]): Map[String, String] = {
+    assert(args.size == 1, s"$INCREMENTAL_TO_AUTO_TAG needs one parameter: endTagName.")
+
+    val endTagName = args.head.eval().toString
+    Map(CoreOptions.INCREMENTAL_TO_AUTO_TAG.key -> endTagName)
+  }
+}

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala

@@ -19,6 +19,7 @@
 package org.apache.paimon.spark
 
 import org.apache.paimon.hive.TestHiveMetastore
+import org.apache.paimon.table.FileStoreTable
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.SparkConf
@@ -78,6 +79,10 @@ class PaimonHiveTestBase extends PaimonSparkTestBase {
     spark.sql(s"USE $sparkCatalogName")
     spark.sql(s"USE $hiveDbName")
   }
+
+  override def loadTable(tableName: String): FileStoreTable = {
+    loadTable(hiveDbName, tableName)
+  }
 }
 
 object PaimonHiveTestBase {

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala

@@ -38,7 +38,7 @@ import org.scalactic.source.Position
 import org.scalatest.Tag
 
 import java.io.File
-import java.util.TimeZone
+import java.util.{TimeZone, UUID}
 
 import scala.util.Random
 
@@ -48,6 +48,8 @@ class PaimonSparkTestBase
   with WithTableOptions
   with SparkVersionSupport {
 
+  protected lazy val commitUser: String = UUID.randomUUID.toString
+
   protected lazy val fileIO: FileIO = LocalFileIO.create
 
   protected lazy val tempDBDir: File = Utils.createTempDir

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala

@@ -18,10 +18,15 @@
 
 package org.apache.paimon.spark.sql
 
+import org.apache.paimon.data.{BinaryString, GenericRow, Timestamp}
+import org.apache.paimon.manifest.ManifestCommittable
 import org.apache.paimon.spark.PaimonHiveTestBase
 
 import org.apache.spark.sql.{DataFrame, Row}
 
+import java.time.LocalDateTime
+import java.util.Collections
+
 class TableValuedFunctionsTest extends PaimonHiveTestBase {
 
   withPk.foreach {
@@ -91,10 +96,117 @@ class TableValuedFunctionsTest extends PaimonHiveTestBase {
       }
   }
 
+  test("Table Valued Functions: paimon_incremental_between_timestamp") {
+    Seq("paimon", sparkCatalogName, paimonHiveCatalogName).foreach {
+      catalogName =>
+        sql(s"USE $catalogName")
+        val dbName = "test_tvf_db"
+        withDatabase(dbName) {
+          sql(s"CREATE DATABASE $dbName")
+          withTable("t") {
+            sql(s"USE $dbName")
+            sql("CREATE TABLE t (id INT) USING paimon")
+
+            sql("INSERT INTO t VALUES 1")
+            Thread.sleep(100)
+            val t1 = System.currentTimeMillis()
+            sql("INSERT INTO t VALUES 2")
+            Thread.sleep(100)
+            val t2 = System.currentTimeMillis()
+            sql("INSERT INTO t VALUES 3")
+            sql("INSERT INTO t VALUES 4")
+            Thread.sleep(100)
+            val t3 = System.currentTimeMillis()
+            sql("INSERT INTO t VALUES 5")
+
+            checkAnswer(
+              sql(
+                s"SELECT * FROM paimon_incremental_between_timestamp('t', '$t1', '$t2') ORDER BY id"),
+              Seq(Row(2)))
+            checkAnswer(
+              sql(
+                s"SELECT * FROM paimon_incremental_between_timestamp('$dbName.t', '$t2', '$t3') ORDER BY id"),
+              Seq(Row(3), Row(4)))
+            checkAnswer(
+              sql(
+                s"SELECT * FROM paimon_incremental_between_timestamp('$catalogName.$dbName.t', '$t1', '$t3') ORDER BY id"),
+              Seq(Row(2), Row(3), Row(4)))
+          }
+        }
+    }
+  }
+
+  test("Table Valued Functions: paimon_incremental_to_auto_tag") {
+    withTable("t") {
+      sql("""
+            |CREATE TABLE t (a INT, b STRING) USING paimon
+            |TBLPROPERTIES ('primary-key' = 'a', 'bucket' = '1', 'tag.automatic-creation'='watermark', 'tag.creation-period'='daily')
+            |""".stripMargin)
+
+      val table = loadTable("t")
+      val write = table.newWrite(commitUser)
+      val commit = table.newCommit(commitUser).ignoreEmptyCommit(false)
+
+      write.write(GenericRow.of(1, BinaryString.fromString("a")))
+      var commitMessages = write.prepareCommit(false, 0)
+      commit.commit(
+        new ManifestCommittable(
+          0,
+          utcMills("2024-12-02T10:00:00"),
+          Collections.emptyMap[Integer, java.lang.Long],
+          commitMessages))
+
+      write.write(GenericRow.of(2, BinaryString.fromString("b")))
+      commitMessages = write.prepareCommit(false, 1)
+      commit.commit(
+        new ManifestCommittable(
+          1,
+          utcMills("2024-12-03T10:00:00"),
+          Collections.emptyMap[Integer, java.lang.Long],
+          commitMessages))
+
+      write.write(GenericRow.of(3, BinaryString.fromString("c")))
+      commitMessages = write.prepareCommit(false, 2)
+      commit.commit(
+        new ManifestCommittable(
+          2,
+          utcMills("2024-12-05T10:00:00"),
+          Collections.emptyMap[Integer, java.lang.Long],
+          commitMessages))
+
+      checkAnswer(
+        sql(s"SELECT * FROM paimon_incremental_to_auto_tag('t', '2024-12-01') ORDER BY a"),
+        Seq())
+      checkAnswer(
+        sql(s"SELECT * FROM paimon_incremental_to_auto_tag('t', '2024-12-02') ORDER BY a"),
+        Seq(Row(2, "b")))
+      checkAnswer(
+        sql(s"SELECT * FROM paimon_incremental_to_auto_tag('t', '2024-12-03') ORDER BY a"),
+        Seq())
+      checkAnswer(
+        sql(s"SELECT * FROM paimon_incremental_to_auto_tag('t', '2024-12-04') ORDER BY a"),
+        Seq(Row(3, "c")))
+    }
+  }
+
   private def incrementalDF(tableIdent: String, start: Int, end: Int): DataFrame = {
     spark.read
       .format("paimon")
       .option("incremental-between", s"$start,$end")
       .table(tableIdent)
   }
+
+  private def utcMills(timestamp: String) =
+    Timestamp.fromLocalDateTime(LocalDateTime.parse(timestamp)).getMillisecond
+
+  object GenericRow {
+    def of(values: Any*): GenericRow = {
+      val row = new GenericRow(values.length)
+      values.zipWithIndex.foreach {
+        case (value, index) =>
+          row.setField(index, value)
+      }
+      row
+    }
+  }
 }