wombatu-kun commented on code in PR #20516:
URL: https://github.com/apache/hudi/pull/20516#discussion_r4477235750
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamSourceV2.scala:
##########
@@ -51,6 +51,10 @@ class HoodieStreamSourceV2(sqlContext: SQLContext,
writeTableVersion: HoodieTableVersion)
extends Source with Logging with Serializable with SparkAdapterSupport {
+ // Capture flow-scoped SQLConf once; explicit reader options take precedence.
+ private val recoveryStrategy = HoodieStreamingSourceSafety.strategy(
+ DataSourceOptionsHelper.collectSparkHoodieConfs(sqlContext, parameters))
Review Comment:
`collectSparkHoodieConfs` is the write-path helper and ignores bare
`hoodie.*` session confs, so `SET
hoodie.datasource.read.streaming.recovery.strategy=...` (the key the
rebuild-required error names) has no effect and the query keeps failing. Could
this use the read-path `collectHoodieAndSparkHoodieConfs` instead?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamingSourceSafety.scala:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.streaming
+
+import org.apache.hudi.{DataSourceReadOptions, DefaultSource,
SparkAdapterSupport}
+import org.apache.hudi.common.model.WriteOperationType
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.CleanerUtils
+import org.apache.hudi.exception.HoodieException
+
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.types.StructType
+
+import scala.collection.JavaConverters._
+
+/** Conservative reset and history validation for checkpointed latest-state
streaming consumers. */
+private[streaming] object HoodieStreamingSourceSafety extends
SparkAdapterSupport {
+ private class RebuildRequired(message: String) extends
HoodieException(message)
+
+ def strategy(parameters: Map[String, String]): String = {
+ val config = DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY
+ val value = parameters.getOrElse(config.key(), config.defaultValue())
+ require(Set("FAIL", "READ_UPTO_LATEST_COMMIT",
"READ_LATEST").contains(value),
+ s"Invalid ${config.key()}=$value; expected FAIL, READ_UPTO_LATEST_COMMIT
or READ_LATEST")
+ value
+ }
+
+ def recover(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ schema: => StructType,
+ parameters: Map[String, String],
+ strategy: String,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Option[DataFrame] = {
+ lazy val recoverySchema = schema
+ try {
+ validate(metaClient, start, end, completionTime, resumed)
Review Comment:
On every restart Spark's `MicroBatchExecution.populateStartOffsets`
re-invokes `getBatch` for the last committed batch and discards the result, so
this re-validates an already-delivered range and fails the restart under `FAIL`
once cleaning or archival has passed it, even when the next batch is readable
(and on the first restart after a recovery batch once the strategy goes back to
`FAIL`). Could the check run lazily inside the returned plan instead, so
Spark's discarded call cannot fail?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamingSourceSafety.scala:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.streaming
+
+import org.apache.hudi.{DataSourceReadOptions, DefaultSource,
SparkAdapterSupport}
+import org.apache.hudi.common.model.WriteOperationType
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.CleanerUtils
+import org.apache.hudi.exception.HoodieException
+
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.types.StructType
+
+import scala.collection.JavaConverters._
+
+/** Conservative reset and history validation for checkpointed latest-state
streaming consumers. */
+private[streaming] object HoodieStreamingSourceSafety extends
SparkAdapterSupport {
+ private class RebuildRequired(message: String) extends
HoodieException(message)
+
+ def strategy(parameters: Map[String, String]): String = {
+ val config = DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY
+ val value = parameters.getOrElse(config.key(), config.defaultValue())
+ require(Set("FAIL", "READ_UPTO_LATEST_COMMIT",
"READ_LATEST").contains(value),
+ s"Invalid ${config.key()}=$value; expected FAIL, READ_UPTO_LATEST_COMMIT
or READ_LATEST")
+ value
+ }
+
+ def recover(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ schema: => StructType,
+ parameters: Map[String, String],
+ strategy: String,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Option[DataFrame] = {
+ lazy val recoverySchema = schema
+ try {
+ validate(metaClient, start, end, completionTime, resumed)
+ None
+ } catch {
+ case error: RebuildRequired => strategy match {
+ case "FAIL" => throw error
+ case "READ_LATEST" =>
+ Some(sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession,
+ sqlContext.sparkContext.emptyRDD[InternalRow], recoverySchema,
isStreaming = true))
+ case "READ_UPTO_LATEST_COMMIT" =>
+ // Spark owns the batch end/checkpoint. Never read a newer snapshot
and checkpoint an older end.
+ val writes =
metaClient.reloadActiveTimeline().getCommitsTimeline.filterCompletedInstants()
+ .getInstants.asScala.toSeq
+ def offset(i: HoodieInstant): String = if (completionTime)
i.getCompletionTime else i.requestedTime()
+ val included = writes.filter(i => offset(i) <= end)
+ val bound = included.map(_.requestedTime()).reduceOption((a, b) =>
if (a > b) a else b)
+ if (bound.isEmpty || writes.exists(i => offset(i) > end &&
i.requestedTime() <= bound.get)) {
Review Comment:
In completion-time mode an async compaction or clustering that was requested
before `bound` and finishes after the pinned `end` trips this guard, and since
Spark retries the same `end` the recovery can never succeed on a MOR table with
async compaction. Should table-service instants (compaction, log compaction,
clustering) be exempt here, since they change no rows?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamSourceV2.scala:
##########
@@ -113,62 +117,68 @@ class HoodieStreamSourceV2(sqlContext: SQLContext,
.getOrElse(initialOffsets)
var endOffset = HoodieSourceOffset(end)
- // We update the offsets here since until this point the latest offsets
have been
- // calculated no matter if it is in the expected version.
- // We translate them here, then the rest logic should be intact.
- startOffset =
HoodieSourceOffset(translateCheckpoint(startOffset.offsetCommitTime))
- endOffset =
HoodieSourceOffset(translateCheckpoint(endOffset.offsetCommitTime))
-
- if (startOffset == endOffset) {
- sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession,
- sqlContext.sparkContext.emptyRDD[InternalRow].setName("empty"),
schema, isStreaming = true)
- } else {
- val (startCompletionTime, rangeType) =
getStartCompletionTimeAndRangeType(startOffset)
- if (isCDCQuery) {
- val cdcOptions = Map(
- DataSourceReadOptions.START_COMMIT.key() -> startCompletionTime,
- DataSourceReadOptions.END_COMMIT.key() -> endOffset.offsetCommitTime
- )
- val relation = if (tableType == HoodieTableType.COPY_ON_WRITE) {
- new HoodieCopyOnWriteCDCHadoopFsRelationFactory(
- sqlContext, metaClient, parameters ++ cdcOptions, schemaOption,
isBootstrappedTable, rangeType).build()
- } else {
- new HoodieMergeOnReadCDCHadoopFsRelationFactory(
- sqlContext, metaClient, parameters ++ cdcOptions, schemaOption,
isBootstrappedTable, rangeType).build()
- }
- sparkAdapter.createStreamingDataFrame(sqlContext, relation,
HoodieCDCFileIndex.FULL_CDC_SPARK_SCHEMA)
+ val recovery = if (isCDCQuery) None else
HoodieStreamingSourceSafety.recover(
+ sqlContext, metaClient, schema, parameters, recoveryStrategy,
startOffset.offsetCommitTime,
+ endOffset.offsetCommitTime, completionTime = true, resumed =
start.isDefined)
+ recovery.getOrElse {
+ // We update the offsets here since until this point the latest offsets
have been
+ // calculated no matter if it is in the expected version.
+ // We translate them here, then the rest logic should be intact.
+ startOffset =
HoodieSourceOffset(translateCheckpoint(startOffset.offsetCommitTime))
+ endOffset =
HoodieSourceOffset(translateCheckpoint(endOffset.offsetCommitTime))
+
+ if (startOffset == endOffset) {
+ sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession,
+ sqlContext.sparkContext.emptyRDD[InternalRow].setName("empty"),
schema, isStreaming = true)
} else {
- // Consume the data between (startCommitTime, endCommitTime]
- val incParams = parameters ++ Map(
- DataSourceReadOptions.QUERY_TYPE.key ->
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL,
- DataSourceReadOptions.START_COMMIT.key -> startCompletionTime,
- DataSourceReadOptions.END_COMMIT.key -> endOffset.offsetCommitTime
- )
-
- if (enableFileGroupReader) {
+ val (startCompletionTime, rangeType) =
getStartCompletionTimeAndRangeType(startOffset)
+ if (isCDCQuery) {
+ val cdcOptions = Map(
+ DataSourceReadOptions.START_COMMIT.key() -> startCompletionTime,
+ DataSourceReadOptions.END_COMMIT.key() ->
endOffset.offsetCommitTime
+ )
val relation = if (tableType == HoodieTableType.COPY_ON_WRITE) {
- new
HoodieCopyOnWriteIncrementalHadoopFsRelationFactoryV2(sqlContext, metaClient,
incParams, Option(schema), isBootstrappedTable, rangeType)
- .build()
+ new HoodieCopyOnWriteCDCHadoopFsRelationFactory(
+ sqlContext, metaClient, parameters ++ cdcOptions, schemaOption,
isBootstrappedTable, rangeType).build()
} else {
- new
HoodieMergeOnReadIncrementalHadoopFsRelationFactoryV2(sqlContext, metaClient,
incParams, Option(schema), isBootstrappedTable, rangeType)
- .build()
+ new HoodieMergeOnReadCDCHadoopFsRelationFactory(
+ sqlContext, metaClient, parameters ++ cdcOptions, schemaOption,
isBootstrappedTable, rangeType).build()
}
- sparkAdapter.createStreamingDataFrame(sqlContext, relation, schema)
+ sparkAdapter.createStreamingDataFrame(sqlContext, relation,
HoodieCDCFileIndex.FULL_CDC_SPARK_SCHEMA)
} else {
- val requiredColumns = schema.fields.map(_.name)
- val rdd = tableType match {
- case HoodieTableType.COPY_ON_WRITE =>
- val serDe = HoodieSparkUtils.getCatalystRowSerDe(schema)
- new IncrementalRelationV2(sqlContext, incParams, Some(schema),
metaClient, rangeType)
- .buildScan(requiredColumns)
- .map(serDe.serializeRow)
- case HoodieTableType.MERGE_ON_READ =>
- MergeOnReadIncrementalRelationV2(sqlContext, incParams,
metaClient, Some(schema), rangeType = rangeType)
- .buildScan(requiredColumns, Array.empty[Filter])
- .asInstanceOf[RDD[InternalRow]]
- case _ => throw new IllegalArgumentException(s"UnSupport
tableType: $tableType")
- }
- sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession, rdd,
schema, isStreaming = true)}
+ // Consume the data between (startCommitTime, endCommitTime]
+ val incParams = parameters ++ Map(
+
DataSourceReadOptions.INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN.key() -> "false",
Review Comment:
Forcing `INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN` off also covers a fresh
`earliest` start, where `validate` is skipped, so on a v8 table whose archived
history includes a clustering or insert_overwrite (replaced files already
cleaned) the first batch now fails on a missing file instead of taking the full
scan master used - and that is also the coordinated-rebuild path the docs
prescribe. Could the override apply only when the start offset is not
`INIT_OFFSET`?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamingSourceSafety.scala:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.streaming
+
+import org.apache.hudi.{DataSourceReadOptions, DefaultSource,
SparkAdapterSupport}
+import org.apache.hudi.common.model.WriteOperationType
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.CleanerUtils
+import org.apache.hudi.exception.HoodieException
+
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.types.StructType
+
+import scala.collection.JavaConverters._
+
+/** Conservative reset and history validation for checkpointed latest-state
streaming consumers. */
+private[streaming] object HoodieStreamingSourceSafety extends
SparkAdapterSupport {
+ private class RebuildRequired(message: String) extends
HoodieException(message)
+
+ def strategy(parameters: Map[String, String]): String = {
+ val config = DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY
+ val value = parameters.getOrElse(config.key(), config.defaultValue())
+ require(Set("FAIL", "READ_UPTO_LATEST_COMMIT",
"READ_LATEST").contains(value),
+ s"Invalid ${config.key()}=$value; expected FAIL, READ_UPTO_LATEST_COMMIT
or READ_LATEST")
+ value
+ }
+
+ def recover(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ schema: => StructType,
+ parameters: Map[String, String],
+ strategy: String,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Option[DataFrame] = {
+ lazy val recoverySchema = schema
+ try {
+ validate(metaClient, start, end, completionTime, resumed)
+ None
+ } catch {
+ case error: RebuildRequired => strategy match {
+ case "FAIL" => throw error
+ case "READ_LATEST" =>
+ Some(sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession,
+ sqlContext.sparkContext.emptyRDD[InternalRow], recoverySchema,
isStreaming = true))
+ case "READ_UPTO_LATEST_COMMIT" =>
+ // Spark owns the batch end/checkpoint. Never read a newer snapshot
and checkpoint an older end.
+ val writes =
metaClient.reloadActiveTimeline().getCommitsTimeline.filterCompletedInstants()
+ .getInstants.asScala.toSeq
+ def offset(i: HoodieInstant): String = if (completionTime)
i.getCompletionTime else i.requestedTime()
+ val included = writes.filter(i => offset(i) <= end)
+ val bound = included.map(_.requestedTime()).reduceOption((a, b) =>
if (a > b) a else b)
+ if (bound.isEmpty || writes.exists(i => offset(i) > end &&
i.requestedTime() <= bound.get)) {
+ throw new HoodieException("Cannot reconstruct the recovery
snapshot at the checkpointed batch end. "
+ + "Rebuild downstream tables and checkpoints together.", error)
+ }
+ val options = parameters ++ Map(
+ DataSourceReadOptions.QUERY_TYPE.key() ->
DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL,
+ DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT.key() -> bound.get)
+ val relation = DefaultSource.createRelation(sqlContext, metaClient,
recoverySchema, options)
+ Some(relation match {
+ case fsRelation: HadoopFsRelation =>
sparkAdapter.createStreamingDataFrame(sqlContext, fsRelation, recoverySchema)
+ case _ =>
sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession,
+
sqlContext.baseRelationToDataFrame(relation).queryExecution.toRdd,
recoverySchema, isStreaming = true)
+ })
+ }
+ }
+ }
+
+ def validate(metaClient: HoodieTableMetaClient,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Unit = {
+ if (start != HoodieTimeline.INIT_INSTANT_TS) {
+ validateRetainedRange(metaClient, start, end, completionTime, resumed)
+ }
+ }
+
+ private def validateRetainedRange(metaClient: HoodieTableMetaClient,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Unit = {
+ val timeline = metaClient.reloadActiveTimeline()
+ val completed =
timeline.filterCompletedInstants().getInstants.asScala.toSeq
+ def offset(instant: HoodieInstant): String = {
+ if (completionTime) instant.getCompletionTime else
instant.requestedTime()
+ }
+ def fail(reason: String): Nothing = {
+ throw new RebuildRequired(s"Streaming source rebuild required for
${metaClient.getBasePath}: $reason. "
+ + "Stop downstream consumers and rebuild affected downstream tables
and checkpoints in dependency order. "
+ + s"To explicitly accept recovery risks, set
${DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY.key()} "
+ + "to READ_UPTO_LATEST_COMMIT (snapshot replay) or READ_LATEST (skip
the gap).")
+ }
+
+ val checkpointInstant = completed.find(i => offset(i) == start)
+ if (resumed && checkpointInstant.isEmpty) {
+ fail(s"saved offset $start is absent from the active timeline; history
expired or the source table was recreated")
+ }
+ if (end < start) {
+ fail(s"source offset moved backwards from $start to $end")
+ }
+
+ val firstWrite =
timeline.getWriteTimeline.filterCompletedInstants().firstInstant()
+ val requestedStart = checkpointInstant.map(_.requestedTime())
+ val unreadWrites =
timeline.getCommitsTimeline.filterCompletedInstants().getInstants.asScala
+ .filter(i => offset(i) > start && offset(i) <= end).toSeq
+ val earliestUnread = unreadWrites.map(_.requestedTime()).reduceOption((a,
b) => if (a < b) a else b)
+ val completedByRequestedTime = completed.map(i => i.requestedTime() ->
i).toMap
+ // Usually only new cleans matter. Retain earlier cleans for
late-completing writes,
+ // or a user-specified initial offset whose requested-time boundary is
unknown.
+ timeline.getCleanerTimeline.filterCompletedInstants().getInstants.asScala
+ .filter(clean => offset(clean) > start || requestedStart.isEmpty ||
+ requestedStart.exists(_ < clean.requestedTime()) ||
earliestUnread.exists(_ < clean.requestedTime()))
+ .foreach { clean =>
+ val metadata = CleanerUtils.getCleanerMetadata(metaClient, clean)
+ val retained = metadata.getEarliestCommitToRetain
+ // File-version cleaning does not expose a retention boundary. Refuse
to infer safety.
+ if ((retained == null || retained.isEmpty) &&
metadata.getTotalFilesDeleted > 0 && offset(clean) > start) {
+ fail("the cleaner did not record an earliest retained commit;
KEEP_LATEST_FILE_VERSIONS "
+ + "cannot prove incremental safety after deleting files")
+ }
+ val boundary = Option(retained).filter(_.nonEmpty)
+ val retainedInstant = completedByRequestedTime.get(retained)
+ val expired = boundary.exists { floor =>
+ requestedStart.exists(_ < floor) || earliestUnread.exists(_ < floor)
||
+ (requestedStart.isEmpty && retainedInstant.exists(i => start <
offset(i))) ||
+ (requestedStart.isEmpty && retainedInstant.isEmpty &&
firstWrite.isPresent)
Review Comment:
When `start` is not an exact active offset (a user-specified `startOffset`),
this disjunct fires for any active clean whose recorded floor is already
archived, which the oldest active clean usually is on a table that cleans and
archives routinely, so even a recent start fails. Should an archived floor only
count when `start` precedes the first active instant?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieStreamingSourceSafety.scala:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.streaming
+
+import org.apache.hudi.{DataSourceReadOptions, DefaultSource,
SparkAdapterSupport}
+import org.apache.hudi.common.model.WriteOperationType
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.CleanerUtils
+import org.apache.hudi.exception.HoodieException
+
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.types.StructType
+
+import scala.collection.JavaConverters._
+
+/** Conservative reset and history validation for checkpointed latest-state
streaming consumers. */
+private[streaming] object HoodieStreamingSourceSafety extends
SparkAdapterSupport {
+ private class RebuildRequired(message: String) extends
HoodieException(message)
+
+ def strategy(parameters: Map[String, String]): String = {
+ val config = DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY
+ val value = parameters.getOrElse(config.key(), config.defaultValue())
+ require(Set("FAIL", "READ_UPTO_LATEST_COMMIT",
"READ_LATEST").contains(value),
+ s"Invalid ${config.key()}=$value; expected FAIL, READ_UPTO_LATEST_COMMIT
or READ_LATEST")
+ value
+ }
+
+ def recover(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ schema: => StructType,
+ parameters: Map[String, String],
+ strategy: String,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Option[DataFrame] = {
+ lazy val recoverySchema = schema
+ try {
+ validate(metaClient, start, end, completionTime, resumed)
+ None
+ } catch {
+ case error: RebuildRequired => strategy match {
+ case "FAIL" => throw error
+ case "READ_LATEST" =>
+ Some(sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession,
+ sqlContext.sparkContext.emptyRDD[InternalRow], recoverySchema,
isStreaming = true))
+ case "READ_UPTO_LATEST_COMMIT" =>
+ // Spark owns the batch end/checkpoint. Never read a newer snapshot
and checkpoint an older end.
+ val writes =
metaClient.reloadActiveTimeline().getCommitsTimeline.filterCompletedInstants()
+ .getInstants.asScala.toSeq
+ def offset(i: HoodieInstant): String = if (completionTime)
i.getCompletionTime else i.requestedTime()
+ val included = writes.filter(i => offset(i) <= end)
+ val bound = included.map(_.requestedTime()).reduceOption((a, b) =>
if (a > b) a else b)
+ if (bound.isEmpty || writes.exists(i => offset(i) > end &&
i.requestedTime() <= bound.get)) {
+ throw new HoodieException("Cannot reconstruct the recovery
snapshot at the checkpointed batch end. "
+ + "Rebuild downstream tables and checkpoints together.", error)
+ }
+ val options = parameters ++ Map(
+ DataSourceReadOptions.QUERY_TYPE.key() ->
DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL,
+ DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT.key() -> bound.get)
+ val relation = DefaultSource.createRelation(sqlContext, metaClient,
recoverySchema, options)
+ Some(relation match {
+ case fsRelation: HadoopFsRelation =>
sparkAdapter.createStreamingDataFrame(sqlContext, fsRelation, recoverySchema)
+ case _ =>
sparkAdapter.internalCreateDataFrame(sqlContext.sparkSession,
+
sqlContext.baseRelationToDataFrame(relation).queryExecution.toRdd,
recoverySchema, isStreaming = true)
+ })
+ }
+ }
+ }
+
+ def validate(metaClient: HoodieTableMetaClient,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Unit = {
+ if (start != HoodieTimeline.INIT_INSTANT_TS) {
+ validateRetainedRange(metaClient, start, end, completionTime, resumed)
+ }
+ }
+
+ private def validateRetainedRange(metaClient: HoodieTableMetaClient,
+ start: String,
+ end: String,
+ completionTime: Boolean,
+ resumed: Boolean): Unit = {
+ val timeline = metaClient.reloadActiveTimeline()
+ val completed =
timeline.filterCompletedInstants().getInstants.asScala.toSeq
+ def offset(instant: HoodieInstant): String = {
+ if (completionTime) instant.getCompletionTime else
instant.requestedTime()
+ }
+ def fail(reason: String): Nothing = {
+ throw new RebuildRequired(s"Streaming source rebuild required for
${metaClient.getBasePath}: $reason. "
+ + "Stop downstream consumers and rebuild affected downstream tables
and checkpoints in dependency order. "
+ + s"To explicitly accept recovery risks, set
${DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY.key()} "
+ + "to READ_UPTO_LATEST_COMMIT (snapshot replay) or READ_LATEST (skip
the gap).")
+ }
+
+ val checkpointInstant = completed.find(i => offset(i) == start)
+ if (resumed && checkpointInstant.isEmpty) {
+ fail(s"saved offset $start is absent from the active timeline; history
expired or the source table was recreated")
+ }
+ if (end < start) {
+ fail(s"source offset moved backwards from $start to $end")
+ }
+
+ val firstWrite =
timeline.getWriteTimeline.filterCompletedInstants().firstInstant()
+ val requestedStart = checkpointInstant.map(_.requestedTime())
+ val unreadWrites =
timeline.getCommitsTimeline.filterCompletedInstants().getInstants.asScala
+ .filter(i => offset(i) > start && offset(i) <= end).toSeq
+ val earliestUnread = unreadWrites.map(_.requestedTime()).reduceOption((a,
b) => if (a < b) a else b)
+ val completedByRequestedTime = completed.map(i => i.requestedTime() ->
i).toMap
+ // Usually only new cleans matter. Retain earlier cleans for
late-completing writes,
+ // or a user-specified initial offset whose requested-time boundary is
unknown.
+ timeline.getCleanerTimeline.filterCompletedInstants().getInstants.asScala
+ .filter(clean => offset(clean) > start || requestedStart.isEmpty ||
+ requestedStart.exists(_ < clean.requestedTime()) ||
earliestUnread.exists(_ < clean.requestedTime()))
+ .foreach { clean =>
+ val metadata = CleanerUtils.getCleanerMetadata(metaClient, clean)
+ val retained = metadata.getEarliestCommitToRetain
+ // File-version cleaning does not expose a retention boundary. Refuse
to infer safety.
+ if ((retained == null || retained.isEmpty) &&
metadata.getTotalFilesDeleted > 0 && offset(clean) > start) {
+ fail("the cleaner did not record an earliest retained commit;
KEEP_LATEST_FILE_VERSIONS "
+ + "cannot prove incremental safety after deleting files")
+ }
+ val boundary = Option(retained).filter(_.nonEmpty)
+ val retainedInstant = completedByRequestedTime.get(retained)
+ val expired = boundary.exists { floor =>
+ requestedStart.exists(_ < floor) || earliestUnread.exists(_ < floor)
||
Review Comment:
`requestedStart.exists(_ < floor)` rejects a caught-up stream whose
checkpoint is a long-running compaction or clustering, since its requested time
can fall below a later clean's earliest retained commit even though every
unread write is still retained. `earliestUnread.exists(_ < floor)` already
covers the files the batch reads, so can this disjunct be dropped?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -651,6 +651,163 @@ 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(strategy: Option[String] = None):
org.apache.spark.sql.execution.streaming.Source = {
+ val options = commonOptions + ("path" -> tablePath) ++
strategy.map { value =>
+ DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY.key() ->
value
+ }
+ 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"))
+ var sessionReader: org.apache.spark.sql.execution.streaming.Source =
null
+ withSQLConf(
+ "spark.hoodie.datasource.read.streaming.recovery.strategy" ->
"FAIL") {
Review Comment:
The session conf here is `FAIL`, which is also the default, so this leg
passes even if the source never reads the session conf at all. Could it set
`READ_LATEST` instead (source built inside the block, `getBatch` outside,
asserting an empty batch) so the SQLConf path SDP depends on is actually pinned?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSource.scala:
##########
@@ -651,6 +651,163 @@ 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(strategy: Option[String] = None):
org.apache.spark.sql.execution.streaming.Source = {
+ val options = commonOptions + ("path" -> tablePath) ++
strategy.map { value =>
+ DataSourceReadOptions.STREAMING_READ_RECOVERY_STRATEGY.key() ->
value
+ }
+ 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"))
+ var sessionReader: org.apache.spark.sql.execution.streaming.Source =
null
+ withSQLConf(
+ "spark.hoodie.datasource.read.streaming.recovery.strategy" ->
"FAIL") {
+ assert(source(strategy =
Some("READ_LATEST")).getBatch(Some(saved), resetEnd).isStreaming)
+ sessionReader = source(strategy = None)
+ }
+ val sessionError =
intercept[org.apache.hudi.exception.HoodieException] {
+ sessionReader.getBatch(Some(saved), resetEnd)
+ }
+ assert(sessionError.getMessage.contains("logical reset"))
+ def batchIds(batch: org.apache.spark.sql.DataFrame): Seq[String] = {
+ val idIndex = batch.schema.fieldIndex("id")
+ batch.queryExecution.logical.collectFirst {
+ case relation:
org.apache.spark.sql.execution.datasources.LogicalRelation =>
+
spark.sqlContext.baseRelationToDataFrame(relation.relation).select("id")
+ .collect().map(_.getString(0)).toSeq.sorted
+ case relation: org.apache.spark.sql.execution.LogicalRDD =>
+
relation.rdd.collect().map(_.getUTF8String(idIndex).toString).toSeq.sorted
+ }.get
+ }
+ val snapshotReader = source(Some("READ_UPTO_LATEST_COMMIT"))
+ assert(batchIds(snapshotReader.getBatch(Some(saved), resetEnd)) ==
Seq("3"))
+ addData(tablePath, Seq(("5", "after-recovery", "50", "005")))
+ // Recovery must remain bounded by Spark's end even if another
commit has arrived.
+ assert(batchIds(snapshotReader.getBatch(Some(saved), resetEnd)) ==
Seq("3"))
+ assert(batchIds(snapshotReader.getBatch(Some(resetEnd),
snapshotReader.getOffset.get)) == Seq("5"))
+ val skipped = source(Some("READ_LATEST")).getBatch(Some(saved),
resetEnd)
+ assert(skipped.queryExecution.logical.collectFirst {
+ case relation: org.apache.spark.sql.execution.LogicalRDD =>
relation.rdd.count()
+ }.contains(0L))
+ intercept[IllegalArgumentException] { source(Some("invalid")) }
+ // A fresh checkpoint rebuilds from the current snapshot instead of
rejecting old resets.
+ assert(source().getBatch(None, resetEnd).isStreaming)
+
+ val fs = new
Path(tablePath).getFileSystem(spark.sessionState.newHadoopConf())
+ val savedTime =
org.apache.spark.sql.hudi.streaming.HoodieSourceOffset(resetEnd).offsetCommitTime
+ val completedReset =
metaClient.reloadActiveTimeline().filterCompletedInstants().getInstantsAsStream
+ .filter(i => (if (completionTime) i.getCompletionTime else
i.requestedTime()) == savedTime)
+ .findFirst().get()
+ val resetFile = new Path(metaClient.getTimelinePath.toString,
+ metaClient.getInstantFileNameGenerator.getFileName(completedReset))
+ assert(fs.delete(resetFile, false))
+ val expiredError =
intercept[org.apache.hudi.exception.HoodieException] {
+ source().getBatch(Some(resetEnd), resetEnd)
+ }
+ assert(expiredError.getMessage.contains("history expired"))
+ fs.delete(new Path(tablePath), true)
+ initialize()
+ addData(tablePath, Seq(("4", "recreated", "40", "004")))
+ val recreationError =
intercept[org.apache.hudi.exception.HoodieException] {
+ source().getBatch(Some(saved), source().getOffset.get)
+ }
+ assert(recreationError.getMessage.contains("rebuild required"))
+ assert(recreationError.getMessage.contains("recreated"))
+ val recreatedReader = source(Some("READ_UPTO_LATEST_COMMIT"))
+ assert(batchIds(recreatedReader.getBatch(Some(saved),
recreatedReader.getOffset.get)) == Seq("4"))
+ }
+ }
+ }
+ }
+
+ Seq(false, true).foreach { unknownBoundary =>
+ test(s"guarded source rejects cleaned history:
unknownBoundary=$unknownBoundary") {
Review Comment:
Every clean in these tests is hand-built and meant to fail, so nothing
checks that a caught-up resumed batch still passes after a real clean
(`expired` hard-wired to `true` would survive). Could one case run the real
cleaner (e.g. `hoodie.clean.commits.retained=2` with repeated updates) and
assert the resumed rows?
--
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]
