shangxinli commented on code in PR #20516: URL: https://github.com/apache/hudi/pull/20516#discussion_r4484661825
########## 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) + } + if (expired) { + fail(s"history after offset $start has been cleaned (earliest retained commit $retained)") + } + } + + completed.filter(i => i.getAction == HoodieTimeline.REPLACE_COMMIT_ACTION && Review Comment: minor: this replace-commit scan has no upper bound at end, so a non-cluster logical reset far beyond the current batch's range still fails the batch instead of only ones within it. ########## 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) { Review Comment: nit: this only catches a checkpoint that's completely gone from the timeline; a checkpoint that's technically present but past the retained history relies on the cleaner check below catching it, not on an explicit range check here. -- 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]
