nsivabalan commented on code in PR #20516:
URL: https://github.com/apache/hudi/pull/20516#discussion_r4474671889
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -651,6 +651,130 @@ class TestStreamingSource extends StreamTest {
)
}
+ Seq(COPY_ON_WRITE, MERGE_ON_READ).foreach { tableType =>
+ Seq(false, true).foreach { completionTime =>
+ test(s"guarded source rejects resets and recreation: $tableType
completionTime=$completionTime") {
+ withTempDir { dir =>
+ val tablePath = s"${dir.getCanonicalPath}/guarded"
+ val conf =
HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf())
+ def initialize(): HoodieTableMetaClient =
HoodieTableMetaClient.newTableBuilder()
+
.setTableType(tableType).setTableName("guarded").setRecordKeyFields("id")
+ .setOrderingFields("ts").initTable(conf, tablePath)
+ val metaClient = initialize()
+ addData(tablePath, Seq(("1", "old", "10", "001")))
+ def source(enabled: Option[Boolean] = Some(true)):
org.apache.spark.sql.execution.streaming.Source = {
+ val options = commonOptions + ("path" -> tablePath) ++ enabled.map
{ value =>
+ DataSourceReadOptions.STREAMING_READ_FAIL_ON_DATA_LOSS.key() ->
value.toString
+ }
+ val checkpoint = s"${dir.getCanonicalPath}/checkpoint"
+ if (completionTime) {
+ new
org.apache.spark.sql.hudi.streaming.HoodieStreamSourceV2(spark.sqlContext,
+ metaClient, checkpoint, None, options,
+
org.apache.spark.sql.hudi.streaming.HoodieEarliestOffsetRangeLimit,
+ HoodieTableVersion.current())
+ } else {
+ new
org.apache.spark.sql.hudi.streaming.HoodieStreamSourceV1(spark.sqlContext,
+ metaClient, checkpoint, None, options,
+
org.apache.spark.sql.hudi.streaming.HoodieEarliestOffsetRangeLimit,
+ HoodieTableVersion.current())
+ }
+ }
+ val reader = source()
+ val saved = reader.getOffset.get
+ addData(tablePath, Seq(("2", "new", "20", "002")),
enableInlineCluster = true)
+ assert(reader.getBatch(Some(saved),
reader.getOffset.get).isStreaming)
+
assert(metaClient.reloadActiveTimeline().getCompletedReplaceTimeline.countInstants()
> 0)
+ val beforeResetEnd = reader.getOffset.get
+ Seq(("3", "replacement", "30", "003")).toDF(columns:
_*).write.format("hudi")
+ .options(commonOptions).option(TBL_NAME.key, "guarded")
+ .option("hoodie.datasource.write.operation",
"insert_overwrite_table")
+ .mode(SaveMode.Append).save(tablePath)
+ val resetEnd = reader.getOffset.get
+ val resetError =
intercept[org.apache.hudi.exception.HoodieException] {
+ reader.getBatch(Some(saved), resetEnd)
+ }
+ assert(resetError.getMessage.contains("rebuild required"))
+ assert(resetError.getMessage.contains("logical reset"))
+ val concurrentResetError =
intercept[org.apache.hudi.exception.HoodieException] {
+ reader.getBatch(Some(saved), beforeResetEnd)
+ }
+ assert(concurrentResetError.getMessage.contains("logical reset"))
+ val sessionReader = withSQLConf(
Review Comment:
On Spark 3.5, `withSQLConf` returns `Unit`, so `sessionReader.getBatch(...)`
below fails to compile (`value getBatch is not a member of Unit`). That's the
red CI. Assign the reader to a `var` inside the block, or set and unset the
conf manually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
