diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/FlussSparkSessionExtensions.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/FlussSparkSessionExtensions.scala index dc12ed5034..5c3faf8638 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/FlussSparkSessionExtensions.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/FlussSparkSessionExtensions.scala @@ -17,7 +17,8 @@ package org.apache.fluss.spark -import org.apache.fluss.spark.catalyst.analysis.FlussProcedureResolver +import org.apache.fluss.spark.catalyst.analysis.{FlussProcedureResolver, FlussTableValuedFunctionResolver} +import org.apache.fluss.spark.catalyst.plans.logical.FlussTableValuedFunctions import org.apache.fluss.spark.execution.FlussStrategy import org.apache.spark.sql.SparkSessionExtensions @@ -32,6 +33,14 @@ class FlussSparkSessionExtensions extends (SparkSessionExtensions => Unit) { // analyzer extensions extensions.injectResolutionRule(spark => FlussProcedureResolver(spark)) + extensions.injectResolutionRule(spark => FlussTableValuedFunctionResolver(spark)) + + // table function extensions + FlussTableValuedFunctions.supportedFnNames.foreach { + fnName => + extensions.injectTableFunction( + FlussTableValuedFunctions.getTableValueFunctionInjection(fnName)) + } // planner extensions extensions.injectPlannerStrategy(spark => FlussStrategy(spark)) diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkFlussConf.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkFlussConf.scala index aac6a698da..aaa8ca1d75 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkFlussConf.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkFlussConf.scala @@ -37,6 +37,13 @@ object SparkFlussConf { val FULL, EARLIEST, LATEST, TIMESTAMP = Value } + object TimestampOutOfRangeMode extends Enumeration { + val ERROR, ADJUST = Value + } + + /** Reserved value of [[SCAN_INCREMENTAL_END_TIMESTAMP]] meaning "the latest committed data". */ + val END_TIMESTAMP_LATEST = "latest" + val SCAN_START_UP_MODE: ConfigOption[String] = ConfigBuilder .key("scan.startup.mode") @@ -44,6 +51,40 @@ object SparkFlussConf { .defaultValue(StartUpMode.FULL.toString) .withDescription("The start up mode when read Fluss table.") + val SCAN_INCREMENTAL_START_TIMESTAMP: ConfigOption[String] = + ConfigBuilder + .key("scan.incremental.start.timestamp") + .stringType() + .noDefaultValue() + .withDescription( + "Enables an incremental (time-range) batch read and sets the inclusive lower bound of " + + "the window. Accepts either epoch milliseconds (e.g. '1678883047356') or a " + + "'yyyy-MM-dd HH:mm:ss' datetime string (e.g. '2023-12-09 23:09:12') interpreted in " + + "the Spark session time zone. Batch read only; it has no effect on streaming reads.") + + val SCAN_INCREMENTAL_END_TIMESTAMP: ConfigOption[String] = + ConfigBuilder + .key("scan.incremental.end.timestamp") + .stringType() + .defaultValue(END_TIMESTAMP_LATEST) + .withDescription( + "The exclusive upper bound of an incremental (time-range) batch read, yielding a " + + "left-closed right-open '[start, end)' window. 'latest' (default) stops at the " + + "latest committed data captured at planning time; otherwise accepts epoch " + + "milliseconds or a 'yyyy-MM-dd HH:mm:ss' datetime string interpreted in the Spark " + + "session time zone. Only honored when 'scan.incremental.start.timestamp' is set.") + + val SCAN_INCREMENTAL_TIMESTAMP_OUT_OF_RANGE: ConfigOption[String] = + ConfigBuilder + .key("scan.incremental.timestamp.out-of-range") + .stringType() + .defaultValue(TimestampOutOfRangeMode.ERROR.toString) + .withDescription( + "Behavior when 'scan.incremental.start.timestamp' precedes the earliest data still " + + "retained by Fluss (bounded by 'table.log.ttl'). 'error' (default): fail fast so a " + + "truncated window is never returned silently. 'adjust': clamp the start to the " + + "earliest retained offset and read from there.") + val SCAN_POLL_TIMEOUT: ConfigOption[Duration] = ConfigBuilder .key("scan.poll.timeout") diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalyst/analysis/FlussTableValuedFunctionResolver.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalyst/analysis/FlussTableValuedFunctionResolver.scala new file mode 100644 index 0000000000..62f0faab69 --- /dev/null +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalyst/analysis/FlussTableValuedFunctionResolver.scala @@ -0,0 +1,37 @@ +/* + * 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.fluss.spark.catalyst.analysis + +import org.apache.fluss.spark.catalyst.plans.logical.{FlussTableValuedFunctions, FlussTableValueFunction} + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.catalyst.rules.Rule + +/** + * Resolution rule for Fluss table-valued functions. The injected table function builder produces an + * unresolved [[FlussTableValueFunction]]; once its arguments are resolved this rule rewrites it + * into a plain DataSourceV2 relation carrying the derived scan options. + */ +case class FlussTableValuedFunctionResolver(sparkSession: SparkSession) extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsDown { + case func: FlussTableValueFunction if func.args.forall(_.resolved) => + FlussTableValuedFunctions.resolveFlussTableValuedFunction(sparkSession, func) + } +} diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalyst/plans/logical/FlussTableValuedFunctions.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalyst/plans/logical/FlussTableValuedFunctions.scala new file mode 100644 index 0000000000..0adf8b9248 --- /dev/null +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalyst/plans/logical/FlussTableValuedFunctions.scala @@ -0,0 +1,225 @@ +/* + * 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.fluss.spark.catalyst.plans.logical + +import org.apache.fluss.spark.{SparkFlussConf, SparkTable} +import org.apache.fluss.spark.catalyst.plans.logical.FlussTableValuedFunctions._ + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.FunctionIdentifier +import org.apache.spark.sql.catalyst.analysis.FunctionRegistryBase +import org.apache.spark.sql.catalyst.analysis.TableFunctionRegistry.TableFunctionBuilder +import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, ExpressionInfo, RuntimeReplaceable} +import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan} +import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog} +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation +import org.apache.spark.sql.types.{IntegerType, LongType, ShortType, StringType, TimestampNTZType, TimestampType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +import scala.collection.JavaConverters._ +import scala.util.control.NonFatal + +/** + * Fluss table-valued functions (TVFs), usable from pure SQL. + * + * A TVF is only sugar over per-relation scan options: the function arguments are translated into + * the same `scan.*` options the DataFrame API accepts, and the call is then resolved into a plain + * [[DataSourceV2Relation]]. Consequently projection, filter push down and metrics keep working, and + * the options are scoped to the single query instead of leaking through session configuration. + */ +object FlussTableValuedFunctions { + + val INCREMENTAL_BETWEEN_TIMESTAMP = "fluss_incremental_between_timestamp" + + val supportedFnNames: Seq[String] = Seq(INCREMENTAL_BETWEEN_TIMESTAMP) + + private type TableFunctionDescription = + (FunctionIdentifier, ExpressionInfo, TableFunctionBuilder) + + def getTableValueFunctionInjection(fnName: String): TableFunctionDescription = { + val (info, builder) = fnName match { + case INCREMENTAL_BETWEEN_TIMESTAMP => + FunctionRegistryBase.build[IncrementalBetweenTimestamp](fnName, since = None) + case _ => + throw new IllegalArgumentException( + s"Function $fnName isn't a supported Fluss table valued function.") + } + (FunctionIdentifier(fnName), info, builder) + } + + /** + * Resolves a Fluss TVF call into a [[DataSourceV2Relation]] over the referenced Fluss table, with + * the function arguments translated into scan options. + */ + def resolveFlussTableValuedFunction( + spark: SparkSession, + tvf: FlussTableValueFunction): LogicalPlan = { + val args = tvf.args + val sessionState = spark.sessionState + val catalogManager = sessionState.catalogManager + + if (args.isEmpty) { + throw new IllegalArgumentException( + s"${tvf.fnName} requires a table identifier as its first argument.") + } + + // Parse the remaining arguments first so that an argument error is reported without depending + // on the referenced table being resolvable. + val options = tvf.parseArgs(args.tail) + + val tableArg = args.head.eval() + if (tableArg == null) { + throw new IllegalArgumentException( + s"The first argument of ${tvf.fnName} must be a non-null table identifier.") + } + val tableIdentifier = tableArg.toString + + val (catalogName, namespace, tableName) = + sessionState.sqlParser.parseMultipartIdentifier(tableIdentifier) 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 IllegalArgumentException( + s"Invalid table identifier '$tableIdentifier' for ${tvf.fnName}. Expected " + + "'table', 'database.table' or 'catalog.database.table'.") + } + + val catalogPlugin = catalogManager.catalog(catalogName) + if (!catalogPlugin.isInstanceOf[TableCatalog]) { + throw new IllegalArgumentException( + s"${tvf.fnName} requires a table catalog, but catalog '$catalogName' is " + + s"${catalogPlugin.getClass.getName}.") + } + val tableCatalog = catalogPlugin.asInstanceOf[TableCatalog] + val ident = Identifier.of(Array(namespace), tableName) + val table = tableCatalog.loadTable(ident) + if (!table.isInstanceOf[SparkTable]) { + throw new IllegalArgumentException( + s"${tvf.fnName} only supports Fluss tables, but '$catalogName.$namespace.$tableName' is " + + s"backed by ${table.getClass.getName}.") + } + + DataSourceV2Relation.create( + table, + Some(tableCatalog), + Some(ident), + new CaseInsensitiveStringMap(options.asJava)) + } + + /** + * Normalizes a timestamp argument to the string form accepted by the `scan.incremental.*` + * timestamp options. + * + * A STRING argument is passed through untouched, so both epoch milliseconds and + * `yyyy-MM-dd HH:mm:ss` keep being interpreted by the option layer. Integral arguments are epoch + * milliseconds. TIMESTAMP arguments are converted from Spark's internal microseconds, otherwise a + * `TIMESTAMP '...'` literal would silently be read as epoch milliseconds. + * + * Any constant expression is accepted, e.g. `CAST(unix_timestamp() * 1000 AS STRING)` or + * `date_format(now() - INTERVAL 1 HOUR, 'yyyy-MM-dd HH:mm:ss')`. + */ + private[logical] def toTimestampOptionValue(fnName: String, expr: Expression): String = { + // `RuntimeReplaceable` expressions (such as the `-` in `now() - INTERVAL 1 HOUR`) only become + // evaluable once the optimizer's ReplaceExpressions rule rewrites them, which has not happened + // yet while the analyzer resolves this function. Apply the same rewrite bottom-up here. + val evaluable = expr.transformUp { case r: RuntimeReplaceable => r.replacement } + + val value = + try { + evaluable.eval() + } catch { + case NonFatal(e) => + throw new IllegalArgumentException( + s"Failed to evaluate the timestamp argument '${expr.sql}' of $fnName. It must be a " + + "constant expression; literals and datetime functions such as now() or " + + "unix_timestamp() are supported, references to table columns are not.", + e + ) + } + if (value == null) { + throw new IllegalArgumentException(s"Timestamp arguments of $fnName must not be null.") + } + evaluable.dataType match { + case StringType => value.toString + case ShortType | IntegerType | LongType => value.toString + case TimestampType | TimestampNTZType => (value.asInstanceOf[Long] / 1000L).toString + case other => + throw new IllegalArgumentException( + s"Unsupported timestamp argument type $other for $fnName. Use a STRING (epoch " + + "milliseconds or 'yyyy-MM-dd HH:mm:ss'), an integral epoch milliseconds value, or a " + + "TIMESTAMP.") + } + } +} + +/** + * An unresolved Fluss table-valued function. + * + * @param fnName + * one of [[FlussTableValuedFunctions.supportedFnNames]]. + */ +abstract class FlussTableValueFunction(val fnName: String) extends LeafNode { + + override def output: Seq[Attribute] = Nil + + override lazy val resolved = false + + val args: Seq[Expression] + + /** Translates the arguments following the table identifier into Fluss scan options. */ + def parseArgs(argsWithoutTable: Seq[Expression]): Map[String, String] +} + +/** + * Plan for [[FlussTableValuedFunctions.INCREMENTAL_BETWEEN_TIMESTAMP]]. + * + * Usage: + * - `fluss_incremental_between_timestamp(table, startTimestamp, endTimestamp)` + * - `fluss_incremental_between_timestamp(table, startTimestamp)` reads up to the latest data + * + * The window is left-closed and right-open, `[start, end)`, on the record commit timestamp. + */ +case class IncrementalBetweenTimestamp(override val args: Seq[Expression]) + extends FlussTableValueFunction(INCREMENTAL_BETWEEN_TIMESTAMP) { + + override def parseArgs(argsWithoutTable: Seq[Expression]): Map[String, String] = { + if (argsWithoutTable.size != 1 && argsWithoutTable.size != 2) { + throw new IllegalArgumentException( + s"$INCREMENTAL_BETWEEN_TIMESTAMP needs a table identifier followed by a startTimestamp " + + s"and an optional endTimestamp, e.g. " + + s"$INCREMENTAL_BETWEEN_TIMESTAMP('db.t', '2026-01-01 00:00:00', '2026-01-01 01:00:00'). " + + s"Got ${argsWithoutTable.size + 1} arguments.") + } + + val start = toTimestampOptionValue(INCREMENTAL_BETWEEN_TIMESTAMP, argsWithoutTable.head) + val startOptions = + Map(SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP.key() -> start) + + // The end bound is always written explicitly so the call stays self-contained: options take + // precedence over session configuration, which may still hold a stale end timestamp. + if (argsWithoutTable.size == 2) { + val end = toTimestampOptionValue(INCREMENTAL_BETWEEN_TIMESTAMP, argsWithoutTable.last) + startOptions + (SparkFlussConf.SCAN_INCREMENTAL_END_TIMESTAMP.key() -> end) + } else { + startOptions + + (SparkFlussConf.SCAN_INCREMENTAL_END_TIMESTAMP.key() -> SparkFlussConf.END_TIMESTAMP_LATEST) + } + } +} diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala index 6351dba48f..0a40f5bdd2 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala @@ -68,7 +68,7 @@ abstract class FlussMicroBatchStream( FlussOffsetInitializers.startOffsetsInitializer(options, flussConfig) val stoppingOffsetsInitializer: OffsetsInitializer = - FlussOffsetInitializers.stoppingOffsetsInitializer(false, options, flussConfig) + FlussOffsetInitializers.stoppingOffsetsInitializer(false, options) protected def projection: Array[Int] = FlussScanBuilder.projectionOf(tableInfo, Some(readSchema)) diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussOffsetInitializers.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussOffsetInitializers.scala index 1f0a8806ae..2745f8ad11 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussOffsetInitializers.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussOffsetInitializers.scala @@ -18,20 +18,90 @@ package org.apache.fluss.spark.read import org.apache.fluss.client.initializer.{NoStoppingOffsetsInitializer, OffsetsInitializer} -import org.apache.fluss.config.Configuration +import org.apache.fluss.config.{ConfigOption, Configuration} import org.apache.fluss.spark.SparkFlussConf +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.util.CaseInsensitiveStringMap +import java.time.{LocalDateTime, ZoneId} +import java.time.format.DateTimeFormatter + object FlussOffsetInitializers { + + private val DATE_TIME_FORMATTER: DateTimeFormatter = + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss") + + /** + * Whether an incremental (time-range) batch read is requested, i.e. + * `scan.incremental.start.timestamp` is set on the relation being scanned. + * + * The `scan.incremental.*` options are read from the per-query scan options only — set by the + * `fluss_incremental_between_timestamp` table-valued function or `DataFrameReader.option` — and + * deliberately not from session configuration, so a window can never leak into another query. + * Streaming reads ignore them. + */ + def isIncrementalRead(options: CaseInsensitiveStringMap): Boolean = { + incrementalOption(options, SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP).isDefined + } + + /** + * Whether a resolved start offset predates the data Fluss still retains for a bucket. A bucket + * whose earliest offset is still 0 has dropped nothing and is never flagged. + */ + def isBeforeRetention(startOffset: Long, earliestOffset: Long): Boolean = + earliestOffset > 0 && startOffset <= earliestOffset + + /** + * Rejects a start offset that predates the earliest retained data (see [[isBeforeRetention]]), so + * a truncated window is never returned silently. Callers must pass a concrete earliest offset, + * i.e. from a retriever built with `fetchEarliestOffset = true`. + */ + def requireStartWithinRetention( + tableDescription: String, + partitionName: String, + bucketId: Int, + startOffset: Long, + earliestOffset: Long): Unit = { + if (isBeforeRetention(startOffset, earliestOffset)) { + val partitionDesc = if (partitionName != null) s" partition '$partitionName'" else "" + throw new IllegalArgumentException( + s"The requested start timestamp resolves to log offset $startOffset for bucket " + + s"$bucketId$partitionDesc of table $tableDescription, which is at or before the " + + s"earliest retained offset $earliestOffset. The requested time range exceeds Fluss " + + s"retention (table.log.ttl); narrow the time range or increase table.log.ttl.") + } + } + + /** + * Whether a start timestamp preceding the earliest retained data fails fast (default) instead of + * being clamped to that offset. Controlled by `scan.incremental.timestamp.out-of-range`. + */ + def failOnTimestampOutOfRange(options: CaseInsensitiveStringMap): Boolean = { + val mode = + incrementalOption( + options, + SparkFlussConf.SCAN_INCREMENTAL_TIMESTAMP_OUT_OF_RANGE).get.toUpperCase + SparkFlussConf.TimestampOutOfRangeMode.withName(mode) == + SparkFlussConf.TimestampOutOfRangeMode.ERROR + } + + /** + * Start offsets of an incremental batch read, resolved from `scan.incremental.start.timestamp`. + * Requires that option to be set. + */ + def incrementalStartOffsetsInitializer(options: CaseInsensitiveStringMap): OffsetsInitializer = + OffsetsInitializer.timestamp( + requiredTimestamp(options, SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP)) + + /** + * Start offsets of a streaming read, driven by `scan.startup.mode`. Batch reads ignore this + * option (see [[incrementalStartOffsetsInitializer]]). + */ def startOffsetsInitializer( options: CaseInsensitiveStringMap, flussConfig: Configuration): OffsetsInitializer = { - val startupMode = options - .getOrDefault( - SparkFlussConf.SCAN_START_UP_MODE.key(), - flussConfig.get(SparkFlussConf.SCAN_START_UP_MODE)) - .toUpperCase + val startupMode = resolveStartupMode(options, flussConfig).toUpperCase SparkFlussConf.StartUpMode.withName(startupMode) match { case SparkFlussConf.StartUpMode.EARLIEST => OffsetsInitializer.earliest() @@ -39,18 +109,88 @@ object FlussOffsetInitializers { case SparkFlussConf.StartUpMode.LATEST => OffsetsInitializer.latest() case _ => throw new IllegalArgumentException( - s"Unsupported scan start up mode: ${options.get(SparkFlussConf.SCAN_START_UP_MODE.key())}") + s"Unsupported scan start up mode: " + + s"${resolveStartupMode(options, flussConfig)}. Supported values are 'full', " + + s"'earliest' and 'latest'. For a time-range batch read set " + + s"'${SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP.key()}' instead.") } } def stoppingOffsetsInitializer( isBatch: Boolean, - options: CaseInsensitiveStringMap, - flussConfig: Configuration): OffsetsInitializer = { - if (isBatch) { + options: CaseInsensitiveStringMap): OffsetsInitializer = { + if (!isBatch) { + new NoStoppingOffsetsInitializer() + } else if (!isIncrementalRead(options)) { + // A plain batch read stops at the latest committed data; an end timestamp alone must not + // truncate it. OffsetsInitializer.latest() } else { - new NoStoppingOffsetsInitializer() + val end = + incrementalOption(options, SparkFlussConf.SCAN_INCREMENTAL_END_TIMESTAMP).getOrElse("").trim + if ( + end.isEmpty || + end.equalsIgnoreCase(SparkFlussConf.END_TIMESTAMP_LATEST) + ) { + OffsetsInitializer.latest() + } else { + OffsetsInitializer.timestamp( + parseTimestamp(end.trim, SparkFlussConf.SCAN_INCREMENTAL_END_TIMESTAMP.key())) + } + } + } + + /** + * Reads a `scan.incremental.*` option from the scan options, falling back to its default. A blank + * value counts as unset, so a whitespace-only start timestamp never enables an incremental read. + */ + private def incrementalOption( + options: CaseInsensitiveStringMap, + option: ConfigOption[String]): Option[String] = + Option(options.getOrDefault(option.key(), option.defaultValue())).filter(_.trim.nonEmpty) + + private def resolveStartupMode( + options: CaseInsensitiveStringMap, + flussConfig: Configuration): String = + options.getOrDefault( + SparkFlussConf.SCAN_START_UP_MODE.key(), + flussConfig.get(SparkFlussConf.SCAN_START_UP_MODE)) + + private def requiredTimestamp( + options: CaseInsensitiveStringMap, + option: ConfigOption[String]): Long = { + val value = incrementalOption(options, option) + if (value.getOrElse("").isEmpty) { + throw new IllegalArgumentException( + s"'${option.key()}' must not be empty. Provide epoch milliseconds or a " + + s"'yyyy-MM-dd HH:mm:ss' timestamp.") + } + parseTimestamp(value.get.trim, option.key()) + } + + /** + * Parses a timestamp option value to epoch milliseconds: a purely numeric string is epoch + * milliseconds, otherwise it is parsed as 'yyyy-MM-dd HH:mm:ss' in the Spark session time zone. + */ + private def parseTimestamp(timestampStr: String, optionKey: String): Long = { + if (timestampStr.matches("\\d+")) { + timestampStr.toLong + } else { + try { + LocalDateTime + .parse(timestampStr, DATE_TIME_FORMATTER) + .atZone(ZoneId.of(SQLConf.get.sessionLocalTimeZone)) + .toInstant + .toEpochMilli + } catch { + case e: Exception => + throw new IllegalArgumentException( + s"Invalid value for '$optionKey': '$timestampStr'. It should be epoch milliseconds or " + + s"follow the format 'yyyy-MM-dd HH:mm:ss', e.g. '2023-12-09 23:09:12' or " + + s"'1678883047356'.", + e + ) + } } } } diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/SplitPlanner.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/SplitPlanner.scala index b0d4c1fca3..d1eb1d5aa2 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/SplitPlanner.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/SplitPlanner.scala @@ -198,6 +198,32 @@ abstract class AbstractSplitPlanner( .toMap } + /** + * Fail-fast guard for an incremental batch read: rejects a start offset that predates the data + * Fluss still retains (bounded by `table.log.ttl`) instead of silently returning a truncated + * window. Requires a retriever created with `fetchEarliestOffset = true`; otherwise + * `earliestOffsets` returns the EARLIEST_OFFSET sentinel (-2) and the guard is a no-op. + */ + protected def checkTimeRangeWithinRetention( + partitionName: String, + buckets: Seq[Int], + startOffsets: scala.collection.Map[Integer, java.lang.Long], + bucketOffsetsRetriever: BucketOffsetsRetrieverImpl): Unit = { + val earliestOffsets = bucketOffsetsRetriever + .earliestOffsets(partitionName, buckets.map(Integer.valueOf).asJava) + .asScala + buckets.foreach { + bucketId => + val bucket = Integer.valueOf(bucketId) + FlussOffsetInitializers.requireStartWithinRetention( + tablePath.toString, + partitionName, + bucketId, + Long2long(startOffsets(bucket)), + Long2long(earliestOffsets(bucket))) + } + } + /** * Releases the Fluss client connection. Idempotent and null-safe; it never forces the lazily * opened connection into existence, so it is a no-op when no metadata access ever occurred. @@ -215,10 +241,10 @@ abstract class AbstractSplitPlanner( } /** - * Single append (log-table) planner. Probes a readable lake snapshot at construction; if present, - * the plan is a union of lake splits and the Fluss log-tail (from each bucket's snapshotLogOffset - * to committed). If absent, the plan is a pure Fluss log scan from earliest to committed - * (SCAN_START_UP_MODE deliberately ignored — see class scaladoc note below). + * Single append (log-table) planner. Probes a readable lake snapshot at construction; if present + * (and not an incremental read), the plan is a union of lake splits and the Fluss log-tail (from + * each bucket's snapshotLogOffset to committed). If absent, the plan is a pure Fluss log scan from + * earliest to committed (SCAN_START_UP_MODE deliberately ignored — see class scaladoc note below). * * Batch semantics note: start offset is hardcoded to [[OffsetsInitializer.full]] instead of * consuming the user-facing SCAN_START_UP_MODE. Rationale — batch reads semantically mean "the full @@ -227,8 +253,10 @@ abstract class AbstractSplitPlanner( * snapshot has no partial-read semantics), which is confusing; (b) with mode=latest and no writes * since planning time, start==stop==tail — an empty range that trips the reader-side * `Invalid offset range` guard. Symmetric "batch = earliest → committed" closes both concerns and - * keeps append/upsert planners aligned. Time-range batch reads should be expressed via predicate - * pushdown on the timestamp column, not startup mode. + * keeps append/upsert planners aligned. A bounded time range is instead requested with the + * batch-only `scan.incremental.start.timestamp` / `scan.incremental.end.timestamp` options, which + * resolve to log offsets identically on append and upsert tables; such an incremental read always + * takes the log-only branch and never unions a lake snapshot. * * `OffsetsInitializer.full()` is chosen over `OffsetsInitializer.earliest()` intentionally: for a * log table the two are semantically equivalent (see OffsetsInitializer.full javadoc), but full() @@ -247,12 +275,26 @@ class AppendPlanner( extends AbstractSplitPlanner(tablePath, tableInfo, flussConfig) with AppendSplitPlanner { - override def hasLakeSnapshot: Boolean = readableLakeSnapshot.isDefined + override def hasLakeSnapshot: Boolean = readableLakeSnapshot.isDefined && !incrementalMode + + private val incrementalMode: Boolean = + FlussOffsetInitializers.isIncrementalRead(options) + + // Whether a start timestamp preceding retained data fails (default) or is clamped to earliest. + private val failOnOutOfRange: Boolean = + FlussOffsetInitializers.failOnTimestampOutOfRange(options) - private val startOffsetsInitializer: OffsetsInitializer = OffsetsInitializer.full() + // An incremental read starts at scan.incremental.start.timestamp, a plain batch read at the + // beginning of the table (see class scaladoc). + private val startOffsetsInitializer: OffsetsInitializer = + if (incrementalMode) { + FlussOffsetInitializers.incrementalStartOffsetsInitializer(options) + } else { + OffsetsInitializer.full() + } override protected val stoppingOffsetsInitializer: OffsetsInitializer = - FlussOffsetInitializers.stoppingOffsetsInitializer(true, options, flussConfig) + FlussOffsetInitializers.stoppingOffsetsInitializer(true, options) // Server-side log filter requires ARROW format. Pushdown already gates this on the log-only // path (never sets pushedPredicate for non-ARROW), but re-checking here keeps the planner @@ -264,15 +306,16 @@ class AppendPlanner( override def plan(): Array[InputPartition] = try { readableLakeSnapshot match { - case Some(snap) => planLakeUnion(snap) - case None => planLogOnly() + // An incremental read never unions a lake snapshot; it reads only Fluss. + case Some(snap) if !incrementalMode => planLakeUnion(snap) + case _ => planLogOnly() } } finally { close() } // --------------------------------------------------------------------------------------------- - // Log-only branch: pure Fluss log scan from earliest → committed with optional range splitting. + // Log-only branch: pure Fluss log scan over [start, stop) with optional range splitting. // --------------------------------------------------------------------------------------------- private def planLogOnly(): Array[InputPartition] = { @@ -281,10 +324,13 @@ class AppendPlanner( if (value > 0) Some(value) else None } - val bucketOffsetsRetrieverImpl = maxRecordsPerPartition match { - case Some(_) => new BucketOffsetsRetrieverImpl(admin, tablePath, true) - case _ => new BucketOffsetsRetrieverImpl(admin, tablePath) - } + // Both the retention guard and the max-records splitter need concrete earliest offsets; + // otherwise the earliest sentinel (-2) is enough. + val bucketOffsetsRetrieverImpl = + new BucketOffsetsRetrieverImpl( + admin, + tablePath, + maxRecordsPerPartition.isDefined || incrementalMode) val buckets = (0 until tableInfo.getNumBuckets).toSeq def splitOffsetRange( @@ -319,13 +365,19 @@ class AppendPlanner( bucketId => val (startOffset, stopOffset) = (startBucketOffsets(bucketId), stoppingBucketOffsets(bucketId)) - val tableBucket = partitionId match { - case Some(pid) => new TableBucket(tableInfo.getTableId, pid, bucketId) - case None => new TableBucket(tableInfo.getTableId, bucketId) - } - maxRecordsPerPartition match { - case Some(maxRecs) => splitOffsetRange(tableBucket, startOffset, stopOffset, maxRecs) - case _ => Seq(FlussAppendInputPartition(tableBucket, startOffset, stopOffset)) + if (startOffset >= stopOffset) { + // Empty range (e.g. a time-range window with no data, or an empty bucket): emit no + // partition so the append reader is not handed an invalid [start, start) range. + Seq.empty[InputPartition] + } else { + val tableBucket = partitionId match { + case Some(pid) => new TableBucket(tableInfo.getTableId, pid, bucketId) + case None => new TableBucket(tableInfo.getTableId, bucketId) + } + maxRecordsPerPartition match { + case Some(maxRecs) => splitOffsetRange(tableBucket, startOffset, stopOffset, maxRecs) + case _ => Seq(FlussAppendInputPartition(tableBucket, startOffset, stopOffset)) + } } }.toArray } @@ -338,14 +390,22 @@ class AppendPlanner( matching .map { partitionInfo => + val partitionName = partitionInfo.getPartitionName val startBucketOffsets = startOffsetsInitializer.getBucketOffsets( - partitionInfo.getPartitionName, + partitionName, buckets.map(Integer.valueOf).asJava, bucketOffsetsRetrieverImpl) val stoppingBucketOffsets = stoppingOffsetsInitializer.getBucketOffsets( - partitionInfo.getPartitionName, + partitionName, buckets.map(Integer.valueOf).asJava, bucketOffsetsRetrieverImpl) + if (incrementalMode && failOnOutOfRange) { + checkTimeRangeWithinRetention( + partitionName, + buckets, + startBucketOffsets.asScala, + bucketOffsetsRetrieverImpl) + } ( partitionInfo.getPartitionId, startBucketOffsets.asScala.map(e => (e._1, Long2long(e._2))), @@ -368,6 +428,13 @@ class AppendPlanner( null, buckets.map(Integer.valueOf).asJava, bucketOffsetsRetrieverImpl) + if (incrementalMode && failOnOutOfRange) { + checkTimeRangeWithinRetention( + null, + buckets, + startBucketOffsets.asScala, + bucketOffsetsRetrieverImpl) + } createPartitions( None, startBucketOffsets.asScala.map(e => (e._1, Long2long(e._2))).toMap, @@ -567,7 +634,9 @@ class AppendPlanner( * partitions. If absent, the plan is a pure Fluss upsert scan derived from kv snapshots + log tail. * * Startup-mode gating has been removed: a batch upsert scan is always full-table regardless of the - * user-facing SCAN_START_UP_MODE setting — same rationale as [[AppendPlanner]]. + * user-facing SCAN_START_UP_MODE setting — same rationale as [[AppendPlanner]]. Setting + * `scan.incremental.start.timestamp` instead yields an incremental read that folds only the + * changelog within the requested window. */ class UpsertPlanner( override val tablePath: TablePath, @@ -580,10 +649,23 @@ class UpsertPlanner( extends AbstractSplitPlanner(tablePath, tableInfo, flussConfig) with UpsertSplitPlanner { - override def hasLakeSnapshot: Boolean = readableLakeSnapshot.isDefined + override def hasLakeSnapshot: Boolean = readableLakeSnapshot.isDefined && !incrementalMode + + private val incrementalMode: Boolean = + FlussOffsetInitializers.isIncrementalRead(options) + + // Whether a start timestamp preceding retained data fails (default) or is clamped to earliest. + private val failOnOutOfRange: Boolean = + FlussOffsetInitializers.failOnTimestampOutOfRange(options) + + // Start offset of an incremental read, resolved from scan.incremental.start.timestamp. Lazy on + // purpose: resolving it requires that option, while a plain batch scan derives its start from kv + // snapshots instead. + private lazy val incrementalStartOffsetsInitializer: OffsetsInitializer = + FlussOffsetInitializers.incrementalStartOffsetsInitializer(options) override protected val stoppingOffsetsInitializer: OffsetsInitializer = - FlussOffsetInitializers.stoppingOffsetsInitializer(true, options, flussConfig) + FlussOffsetInitializers.stoppingOffsetsInitializer(true, options) // Upsert never pushes a server-side log filter (kv+log union semantics require full log tail // to be reconciled with kv snapshots — see FlussUpsertPartitionReader). @@ -592,6 +674,9 @@ class UpsertPlanner( override def plan(): Array[InputPartition] = try { readableLakeSnapshot match { + // An incremental read reads neither the lake nor the kv snapshot; it folds only the Fluss + // changelog within [start, end). + case _ if incrementalMode => planIncrementalLogOnly() case Some(snap) => planLakeUnion(snap) case None => planLogOnly() } @@ -657,6 +742,72 @@ class UpsertPlanner( .toArray } + // --------------------------------------------------------------------------------------------- + // Incremental branch: fold the Fluss changelog within [start, end) per bucket, with no kv + // snapshot and no lake. Emitting snapshotId = -1 makes FlussUpsertPartitionReader skip the + // snapshot and fold only the log range; SortMergeReader drops delete rows, so the output is the + // surviving +I/+U rows (keys inserted or updated in the window; deleted keys excluded). + // --------------------------------------------------------------------------------------------- + + private def planIncrementalLogOnly(): Array[InputPartition] = { + val bucketOffsetsRetriever = new BucketOffsetsRetrieverImpl(admin, tablePath, true) + val buckets = (0 until tableInfo.getNumBuckets).toSeq + + if (tableInfo.isPartitioned) { + val matching = SparkPartitionPredicate.filterPartitions( + tableInfo, + partitionInfos.asScala.toSeq, + partitionPredicate) + matching.flatMap { + partitionInfo => + createIncrementalUpsertPartitions( + partitionInfo.getPartitionName, + Some(partitionInfo.getPartitionId), + buckets, + bucketOffsetsRetriever) + }.toArray + } else { + createIncrementalUpsertPartitions(null, None, buckets, bucketOffsetsRetriever) + } + } + + private def createIncrementalUpsertPartitions( + partitionName: String, + partitionId: Option[Long], + buckets: Seq[Int], + bucketOffsetsRetriever: BucketOffsetsRetrieverImpl): Array[InputPartition] = { + val jBuckets = buckets.map(Integer.valueOf).asJava + val startBucketOffsets = + incrementalStartOffsetsInitializer.getBucketOffsets( + partitionName, + jBuckets, + bucketOffsetsRetriever) + val stoppingBucketOffsets = + stoppingOffsetsInitializer.getBucketOffsets(partitionName, jBuckets, bucketOffsetsRetriever) + if (failOnOutOfRange) { + checkTimeRangeWithinRetention( + partitionName, + buckets, + startBucketOffsets.asScala, + bucketOffsetsRetriever) + } + + val tableId = tableInfo.getTableId + buckets.map { + bucketId => + val tableBucket = partitionId match { + case Some(pid) => new TableBucket(tableId, pid, bucketId) + case None => new TableBucket(tableId, bucketId) + } + FlussUpsertInputPartition( + tableBucket, + -1L, + Long2long(startBucketOffsets.get(Integer.valueOf(bucketId))), + Long2long(stoppingBucketOffsets.get(Integer.valueOf(bucketId)))) + .asInstanceOf[InputPartition] + }.toArray + } + // --------------------------------------------------------------------------------------------- // Lake-union branch: lake splits (upsert view) + Fluss log tail after snapshotLogOffset. // --------------------------------------------------------------------------------------------- diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkTimeRangeTvfTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkTimeRangeTvfTest.scala new file mode 100644 index 0000000000..f588670958 --- /dev/null +++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkTimeRangeTvfTest.scala @@ -0,0 +1,622 @@ +/* + * 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.fluss.spark + +import org.apache.fluss.client.table.Table +import org.apache.fluss.row.{BinaryString, GenericRow} + +import org.apache.spark.sql.Row +import org.assertj.core.api.Assertions.assertThat + +import java.time.{Duration, Instant, ZoneId} +import java.time.format.DateTimeFormatter + +/** + * Verifies the `fluss_incremental_between_timestamp` table-valued function. The window is + * left-closed, right-open `[start, end)` on the record commit timestamp, and the function's options + * are scoped to the single query. + */ +class SparkTimeRangeTvfTest extends FlussSparkTestBase { + + private val TVF = "fluss_incremental_between_timestamp" + + private def createLogTable(name: String): Unit = + sql(s""" + |CREATE TABLE $DEFAULT_DATABASE.$name + |(orderId BIGINT, itemId BIGINT, amount INT, address STRING) + |""".stripMargin) + + /** Truncates to a whole second so a millis value, a datetime string and a TIMESTAMP agree. */ + private def secondAligned(ms: Long): Long = (ms / 1000L) * 1000L + + private def waitPast(ms: Long): Unit = { + while (System.currentTimeMillis() <= ms) { + Thread.sleep(20) + } + } + + private def formatTs(ms: Long): String = + Instant + .ofEpochMilli(ms) + .atZone(ZoneId.of(spark.sessionState.conf.sessionLocalTimeZone)) + .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + + private def fullMessage(t: Throwable): String = { + val sw = new java.io.StringWriter() + t.printStackTrace(new java.io.PrintWriter(sw)) + sw.toString + } + + test("TVF: log table window [t1, t2)") { + withTable("t") { + createLogTable("t") + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES + |(1L, 11L, 101, "a1"), (2L, 12L, 102, "a2")""".stripMargin) + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES + |(3L, 13L, 103, "a3"), (4L, 14L, 104, "a4")""".stripMargin) + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (5L, 15L, 105, "a5")""") + Thread.sleep(200) + + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2') ORDER BY orderId"), + Row(3L, 13L, 103, "a3") :: Row(4L, 14L, 104, "a4") :: Nil) + + // projection and filter still work on top of the TVF relation + checkAnswer( + sql(s"""SELECT address FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2') + |WHERE amount = 104""".stripMargin), + Row("a4") :: Nil) + } + } + + test("TVF: two-argument form reads up to the latest data") { + withTable("t") { + createLogTable("t") + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (1L, 11L, 101, "a1")""") + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (2L, 12L, 102, "a2")""") + Thread.sleep(300) + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (3L, 13L, 103, "a3")""") + Thread.sleep(200) + + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1') ORDER BY orderId"), + Row(2L, 12L, 102, "a2") :: Row(3L, 13L, 103, "a3") :: Nil) + } + } + + private def createPkTable(name: String): Unit = + sql(s""" + |CREATE TABLE $DEFAULT_DATABASE.$name + |(orderId BIGINT, itemId BIGINT, amount INT, address STRING) + |TBLPROPERTIES("primary.key" = "orderId", "bucket.num" = 1) + |""".stripMargin) + + private def createPartitionedPkTable(name: String): Unit = + sql(s""" + |CREATE TABLE $DEFAULT_DATABASE.$name + |(orderId BIGINT, itemId BIGINT, amount INT, address STRING, dt STRING) + |PARTITIONED BY (dt) + |TBLPROPERTIES("primary.key" = "orderId,dt", "bucket.num" = 1) + |""".stripMargin) + + test("TVF: primary key table folds to +I/+U and excludes deletes") { + withTable("t") { + val tablePath = createTablePath("t") + createPkTable("t") + + val writer = loadFlussTable(tablePath).newUpsert().createWriter() + writer.upsert(row(1L, 11L, 101, "a1")).get() + writer.upsert(row(2L, 12L, 102, "a2")).get() + writer.upsert(row(3L, 13L, 103, "a3")).get() + writer.flush() + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + // in-window: update key 2, insert key 4, delete key 1 + writer.upsert(row(2L, 120L, 1002, "a2_upd")).get() + writer.upsert(row(4L, 14L, 104, "a4")).get() + writer.delete(deleteKey(1L)).get() + writer.flush() + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + // after the window + writer.upsert(row(5L, 15L, 105, "a5")).get() + writer.flush() + Thread.sleep(200) + + val table = loadFlussTable(tablePath) + // evidence: the window changelog really contains -U/+U (key 2), +I (key 4), -D (key 1) + val changes = changelogInWindow(table, t1, t2) + assertThat(changes.filter(_._1 == "-U").map(_._2)).isEqualTo(Seq(2L)) + assertThat(changes.filter(_._1 == "+U").map(_._2)).isEqualTo(Seq(2L)) + assertThat(changes.filter(_._1 == "+I").map(_._2)).isEqualTo(Seq(4L)) + assertThat(changes.filter(_._1 == "-D").map(_._2)).isEqualTo(Seq(1L)) + + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2') ORDER BY orderId"), + Row(2L, 120L, 1002, "a2_upd") :: Row(4L, 14L, 104, "a4") :: Nil) + } + } + + test("TVF: primary key table collapses repeated -U/+U updates into the latest value") { + withTable("t") { + val tablePath = createTablePath("t") + createPkTable("t") + + val writer = loadFlussTable(tablePath).newUpsert().createWriter() + // before the window: keys 1-3 inserted + writer.upsert(row(1L, 11L, 101, "a1")).get() + writer.upsert(row(2L, 12L, 102, "a2")).get() + writer.upsert(row(3L, 13L, 103, "a3")).get() + writer.flush() + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + // in-window: -U/+U twice on key 1, -U/+U once on key 2, key 3 untouched + writer.upsert(row(1L, 110L, 1001, "a1_v2")).get() + writer.upsert(row(1L, 111L, 1002, "a1_v3")).get() + writer.upsert(row(2L, 120L, 2001, "a2_v2")).get() + writer.flush() + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + // after the window + writer.upsert(row(1L, 112L, 1003, "a1_v4")).get() + writer.flush() + Thread.sleep(200) + + // evidence: three genuine -U/+U pairs exist in the window changelog (two for key 1, one + // for key 2), so the folding assertions below operate on real -U/+U records + val changes = changelogInWindow(loadFlussTable(tablePath), t1, t2) + assertThat(changes.filter(_._1 == "-U").map(_._2)).isEqualTo(Seq(1L, 1L, 2L)) + assertThat(changes.filter(_._1 == "+U").map(_._2)).isEqualTo(Seq(1L, 1L, 2L)) + + // each updated key appears exactly once, with its last in-window value + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2') ORDER BY orderId"), + Row(1L, 111L, 1002, "a1_v3") :: Row(2L, 120L, 2001, "a2_v2") :: Nil) + + // the two-argument form reads through to the latest state + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1') ORDER BY orderId"), + Row(1L, 112L, 1003, "a1_v4") :: Row(2L, 120L, 2001, "a2_v2") :: Nil) + } + } + + test("TVF: primary key table cancels out +I followed by -D in the window") { + withTable("t") { + val tablePath = createTablePath("t") + createPkTable("t") + + val writer = loadFlussTable(tablePath).newUpsert().createWriter() + // before the window + writer.upsert(row(1L, 11L, 101, "a1")).get() + writer.flush() + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + // in-window: insert key 2 then delete it again (cancels out), insert key 3 (survives) + writer.upsert(row(2L, 12L, 102, "a2")).get() + writer.delete(deleteKey(2L)).get() + writer.upsert(row(3L, 13L, 103, "a3")).get() + writer.flush() + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + // evidence: the window changelog holds +I then -D for key 2, and +I for key 3 + val changes = changelogInWindow(loadFlussTable(tablePath), t1, t2) + assertThat(changes.filter(r => r._1 == "+I" || r._1 == "-D")) + .isEqualTo(Seq(("+I", 2L), ("-D", 2L), ("+I", 3L))) + + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2') ORDER BY orderId"), + Row(3L, 13L, 103, "a3") :: Nil) + } + } + + test("TVF: primary key table keeps a key deleted then re-inserted in the window") { + withTable("t") { + val tablePath = createTablePath("t") + createPkTable("t") + + val writer = loadFlussTable(tablePath).newUpsert().createWriter() + // before the window + writer.upsert(row(1L, 11L, 101, "a1")).get() + writer.upsert(row(2L, 12L, 102, "a2")).get() + writer.flush() + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + // in-window: delete key 1 then re-insert it with new values (-D then +I survives), + // delete key 2 permanently (-D only, excluded) + writer.delete(deleteKey(1L)).get() + writer.upsert(row(1L, 110L, 1001, "a1_new")).get() + writer.delete(deleteKey(2L)).get() + writer.flush() + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + // evidence: the window changelog holds -D then +I for key 1, and -D for key 2 + assertThat(changelogInWindow(loadFlussTable(tablePath), t1, t2)) + .isEqualTo(Seq(("-D", 1L), ("+I", 1L), ("-D", 2L))) + + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2') ORDER BY orderId"), + Row(1L, 110L, 1001, "a1_new") :: Nil) + } + } + + test("TVF: primary key table window containing only -D returns nothing") { + withTable("t") { + val tablePath = createTablePath("t") + createPkTable("t") + + val writer = loadFlussTable(tablePath).newUpsert().createWriter() + writer.upsert(row(1L, 11L, 101, "a1")).get() + writer.upsert(row(2L, 12L, 102, "a2")).get() + writer.flush() + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + writer.delete(deleteKey(1L)).get() + writer.delete(deleteKey(2L)).get() + writer.flush() + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + // evidence: the window changelog holds exactly two -D records, so the empty result below + // reflects genuine delete folding rather than an empty window + val changes = changelogInWindow(loadFlussTable(tablePath), t1, t2) + assertThat(changes.map(_._1)).isEqualTo(Seq("-D", "-D")) + assertThat(changes.map(_._2)).isEqualTo(Seq(1L, 2L)) + + checkAnswer(sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2')"), Nil) + } + } + + test("TVF: partitioned primary key table folds changes across partitions") { + withTable("t_pk_part") { + val tablePath = createTablePath("t_pk_part") + createPartitionedPkTable("t_pk_part") + + val writer = loadFlussTable(tablePath).newUpsert().createWriter() + // before the window: one row per partition + writer.upsert(pkRow(1L, 11L, 101, "a1", "2026-01-01")).get() + writer.upsert(pkRow(2L, 12L, 102, "a2", "2026-01-02")).get() + writer.flush() + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + // in-window: update key 1 (partition 1), permanent delete of key 2 (partition 2), + // insert key 3 then delete it again in partition 1 (cancels out) + writer.upsert(pkRow(1L, 110L, 1001, "a1_upd", "2026-01-01")).get() + writer.delete(deleteKey(2L, "2026-01-02")).get() + writer.upsert(pkRow(3L, 13L, 103, "a3", "2026-01-01")).get() + writer.delete(deleteKey(3L, "2026-01-01")).get() + writer.flush() + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + // after the window + writer.upsert(pkRow(4L, 14L, 104, "a4", "2026-01-01")).get() + writer.flush() + Thread.sleep(200) + + // evidence: the window changelog holds -U/+U (key 1), -D (key 2), +I then -D (key 3); + // the -D comparison sorts first because poll order across partitions is not deterministic + val changes = changelogInWindow(loadFlussTable(tablePath), t1, t2) + assertThat(changes.filter(_._1 == "-U").map(_._2)).isEqualTo(Seq(1L)) + assertThat(changes.filter(_._1 == "+U").map(_._2)).isEqualTo(Seq(1L)) + assertThat(changes.filter(_._1 == "-D").map(_._2).sorted).isEqualTo(Seq(2L, 3L)) + assertThat(changes.filter(_._1 == "+I").map(_._2)).isEqualTo(Seq(3L)) + + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t_pk_part', '$t1', '$t2') ORDER BY orderId"), + Row(1L, 110L, 1001, "a1_upd", "2026-01-01") :: Nil) + + // partition filter on top of the TVF relation + checkAnswer( + sql(s"""SELECT orderId FROM $TVF('$DEFAULT_DATABASE.t_pk_part', '$t1') + |WHERE dt = '2026-01-01' ORDER BY orderId""".stripMargin), + Row(1L) :: Row(4L) :: Nil + ) + } + } + + test("TVF: session-level scan.incremental.* options are ignored") { + withTable("t") { + createLogTable("t") + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (1L, 11L, 101, "a1")""") + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (2L, 12L, 102, "a2")""") + Thread.sleep(200) + + // A stale window in session configuration must not leak into reads: the scan.incremental.* + // options are only honored as per-query scan options (TVF arguments / DataFrameReader). + withSQLConf( + s"spark.sql.fluss.${SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP.key()}" -> + "2000-01-01 00:00:00", + s"spark.sql.fluss.${SparkFlussConf.SCAN_INCREMENTAL_END_TIMESTAMP.key()}" -> + "2000-01-02 00:00:00" + ) { + // A plain batch read still returns the full table. + checkAnswer( + sql(s"SELECT * FROM $DEFAULT_DATABASE.t ORDER BY orderId"), + Row(1L, 11L, 101, "a1") :: Row(2L, 12L, 102, "a2") :: Nil) + + // The TVF window is unaffected by the session values. + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1') ORDER BY orderId"), + Row(2L, 12L, 102, "a2") :: Nil) + } + } + } + + test("TVF: empty window returns no rows") { + withTable("t") { + createLogTable("t") + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (1L, 11L, 101, "a1")""") + Thread.sleep(400) + val ta = System.currentTimeMillis() + Thread.sleep(300) + val tb = System.currentTimeMillis() + Thread.sleep(300) + // written after the [ta, tb) gap, so the window contains no data + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (2L, 12L, 102, "a2")""") + Thread.sleep(200) + + checkAnswer(sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$ta', '$tb')"), Nil) + } + } + + test("TVF: epoch millis, datetime string and TIMESTAMP literal yield the same window") { + withTable("t") { + createLogTable("t") + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (1L, 11L, 101, "a1")""") + Thread.sleep(1500) + val t1 = secondAligned(System.currentTimeMillis()) + waitPast(t1) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (2L, 12L, 102, "a2")""") + Thread.sleep(1500) + val t2 = secondAligned(System.currentTimeMillis()) + waitPast(t2) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (3L, 13L, 103, "a3")""") + Thread.sleep(200) + + val expected = Row(2L, 12L, 102, "a2") :: Nil + + // epoch milliseconds as a string + checkAnswer(sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '$t1', '$t2')"), expected) + // epoch milliseconds as an integral literal + checkAnswer(sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', ${t1}L, ${t2}L)"), expected) + // 'yyyy-MM-dd HH:mm:ss' in the session time zone + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '${formatTs(t1)}', '${formatTs(t2)}')"), + expected) + // TIMESTAMP literals + checkAnswer( + sql(s"""SELECT * FROM $TVF('$DEFAULT_DATABASE.t', + |TIMESTAMP '${formatTs(t1)}', TIMESTAMP '${formatTs(t2)}')""".stripMargin), + expected + ) + } + } + + test("TVF: Spark expressions as timestamp arguments") { + withTable("t") { + createLogTable("t") + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES + |(1L, 11L, 101, "a1"), (2L, 12L, 102, "a2"), (3L, 13L, 103, "a3")""".stripMargin) + // unix_timestamp() has second granularity, so make every row strictly older than the + // truncated "now" to keep the window boundaries deterministic. + Thread.sleep(1300) + + // [now - 1h, now) covers every row written above, as epoch milliseconds + // (unix_timestamp() returns seconds) + checkAnswer( + sql(s"""SELECT * FROM $TVF( + | '$DEFAULT_DATABASE.t', + | CAST((unix_timestamp() - 3600) * 1000 AS STRING), + | CAST(unix_timestamp() * 1000 AS STRING)) ORDER BY orderId""".stripMargin), + Row(1L, 11L, 101, "a1") :: Row(2L, 12L, 102, "a2") :: Row(3L, 13L, 103, "a3") :: Nil + ) + + // the same window as datetime strings + checkAnswer( + sql(s"""SELECT * FROM $TVF( + | '$DEFAULT_DATABASE.t', + | date_format(now() - INTERVAL 1 HOUR, 'yyyy-MM-dd HH:mm:ss'), + | date_format(now(), 'yyyy-MM-dd HH:mm:ss')) ORDER BY orderId""".stripMargin), + Row(1L, 11L, 101, "a1") :: Row(2L, 12L, 102, "a2") :: Row(3L, 13L, 103, "a3") :: Nil + ) + + // [now, latest) excludes them, proving the expression is really evaluated and applied + checkAnswer( + sql(s"""SELECT * FROM $TVF( + | '$DEFAULT_DATABASE.t', + | CAST(unix_timestamp() * 1000 AS STRING))""".stripMargin), + Nil + ) + } + } + + test("TVF: partitioned log table window read") { + withTable("t_part") { + sql(s""" + |CREATE TABLE $DEFAULT_DATABASE.t_part + |(orderId BIGINT, itemId BIGINT, amount INT, dt STRING) + |PARTITIONED BY (dt) + |""".stripMargin) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t_part VALUES + |(1L, 11L, 101, "2026-01-01"), (2L, 12L, 102, "2026-01-02")""".stripMargin) + Thread.sleep(500) + val t1 = System.currentTimeMillis() + Thread.sleep(50) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t_part VALUES + |(3L, 13L, 103, "2026-01-01"), (4L, 14L, 104, "2026-01-02")""".stripMargin) + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + + sql(s"""INSERT INTO $DEFAULT_DATABASE.t_part VALUES (5L, 15L, 105, "2026-01-01")""") + Thread.sleep(200) + + checkAnswer( + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t_part', '$t1', '$t2') ORDER BY orderId"), + Row(3L, 13L, 103, "2026-01-01") :: Row(4L, 14L, 104, "2026-01-02") :: Nil + ) + + // partition filter on top of the TVF relation + checkAnswer( + sql(s"""SELECT orderId FROM $TVF('$DEFAULT_DATABASE.t_part', '$t1', '$t2') + |WHERE dt = '2026-01-01'""".stripMargin), + Row(3L) :: Nil) + } + } + + test("TVF: wrong argument count fails with a usage hint") { + withTable("t") { + createLogTable("t") + sql(s"""INSERT INTO $DEFAULT_DATABASE.t VALUES (1L, 11L, 101, "a1")""") + + // only the table identifier + val tooFew = intercept[Exception] { + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t')").collect() + } + assertThat(fullMessage(tooFew)).contains("endTimestamp") + + // one argument too many + val tooMany = intercept[Exception] { + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t', '1', '2', '3')").collect() + } + assertThat(fullMessage(tooMany)).contains("endTimestamp") + } + } + + test("TVF: unknown table fails") { + val ex = intercept[Exception] { + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.not_exist_tvf_table', '1', '2')").collect() + } + assertThat(fullMessage(ex)).contains("not_exist_tvf_table") + } + + private def row(orderId: Long, itemId: Long, amount: Int, address: String): GenericRow = + GenericRow.of( + Long.box(orderId), + Long.box(itemId), + Int.box(amount), + BinaryString.fromString(address)) + + /** + * Raw changelog records of `table` whose commit timestamp falls inside [start, end), as + * (changeType, orderId) pairs in log order. Used to prove the claimed change types (-U/+U/-D/+I) + * really exist in the window, so the folded-output assertions below cannot pass vacuously. + */ + private def changelogInWindow(table: Table, start: Long, end: Long): Seq[(String, Long)] = { + val scanner = table.newScan().createLogScanner() + try { + if (table.getTableInfo.isPartitioned) { + admin.listPartitionInfos(table.getTableInfo.getTablePath).get().forEach { + pi => scanner.subscribeFromBeginning(pi.getPartitionId, 0) + } + } else { + scanner.subscribeFromBeginning(0) + } + val records = scala.collection.mutable.ArrayBuffer[(String, Long)]() + // Poll until records arrive and a poll comes back empty (all caught up), or the deadline. + // Mirrors FlussSparkTestBase.getRowsWithChangeType: the high watermark may advance in + // stages, so a single early empty poll must not end the scan. + val deadline = System.currentTimeMillis() + 10000 + var hasReceivedAny = false + var done = false + while (!done && System.currentTimeMillis() < deadline) { + val polled = scanner.poll(Duration.ofSeconds(1)) + if (!polled.isEmpty) { + hasReceivedAny = true + polled.forEach { + r => + if (r.timestamp() >= start && r.timestamp() < end) { + records += ((r.getChangeType.shortString(), r.getRow.getLong(0))) + } + } + } else if (hasReceivedAny) { + done = true + } + } + records.toSeq + } finally { + scanner.close() + } + } + + private def pkRow( + orderId: Long, + itemId: Long, + amount: Int, + address: String, + dt: String): GenericRow = + GenericRow.of( + Long.box(orderId), + Long.box(itemId), + Int.box(amount), + BinaryString.fromString(address), + BinaryString.fromString(dt)) + + private def deleteKey(orderId: Long): GenericRow = + GenericRow.of(Long.box(orderId), null, null, null) + + private def deleteKey(orderId: Long, dt: String): GenericRow = + GenericRow.of(Long.box(orderId), null, null, null, BinaryString.fromString(dt)) +} diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeTimeRangeReadTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeTimeRangeReadTest.scala new file mode 100644 index 0000000000..84d9ba9e9e --- /dev/null +++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeTimeRangeReadTest.scala @@ -0,0 +1,151 @@ +/* + * 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.fluss.spark.lake + +import org.apache.fluss.config.{ConfigOptions, Configuration} +import org.apache.fluss.metadata.DataLakeFormat +import org.apache.fluss.spark.SparkConnectorOptions.{BUCKET_NUMBER, PRIMARY_KEY} +import org.apache.fluss.spark.read.{FlussAppendInputPartition, FlussUpsertInputPartition} + +import org.apache.spark.sql.Row + +import java.nio.file.Files + +/** + * Verifies that an incremental (time-range) batch read on a lake-enabled table is forced to the + * log-only branch: even when a readable lake snapshot exists, the plan never unions lake splits and + * never reads the kv/lake snapshot, so only the data still retained in Fluss is returned. The + * result is the `[t1, t2)` window folded per the underlying table type. + */ +abstract class SparkLakeTimeRangeReadTest extends SparkLakeTableReadTestBase { + + private val TVF = "fluss_incremental_between_timestamp" + + test("Spark Lake Read: log table time-range forces log-only (skips lake snapshot)") { + withTable("t_lake_tr_log") { + sql(s""" + |CREATE TABLE $DEFAULT_DATABASE.t_lake_tr_log (id INT, name STRING) + | TBLPROPERTIES ( + | '${ConfigOptions.TABLE_DATALAKE_ENABLED.key()}' = true, + | '${ConfigOptions.TABLE_DATALAKE_FRESHNESS.key()}' = '1s', + | '${BUCKET_NUMBER.key()}' = 1) + |""".stripMargin) + + // group 1 (before the window) -> tiered to lake, but also still retained in Fluss + sql(s"""INSERT INTO $DEFAULT_DATABASE.t_lake_tr_log VALUES (1, "hello"), (2, "world")""") + tierToLake("t_lake_tr_log") + + val t1 = System.currentTimeMillis() + Thread.sleep(50) + // group 2 (inside the window) + sql(s"""INSERT INTO $DEFAULT_DATABASE.t_lake_tr_log VALUES (3, "fluss"), (4, "spark")""") + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + // group 3 (after the window) + sql(s"""INSERT INTO $DEFAULT_DATABASE.t_lake_tr_log VALUES (5, "lake")""") + Thread.sleep(200) + + val df = + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t_lake_tr_log', '$t1', '$t2') ORDER BY id") + val partitions = lakeInputPartitions(df) + assert(partitions.nonEmpty, "expected at least one Fluss log partition") + assert( + partitions.forall(_.isInstanceOf[FlussAppendInputPartition]), + s"time-range read must be log-only (no lake splits), got: ${partitions.mkString(", ")}" + ) + checkAnswer(df, Row(3, "fluss") :: Row(4, "spark") :: Nil) + } + } + + test("Spark Lake Read: pk table time-range forces log-only (skips lake + kv snapshot)") { + withTable("t_lake_tr_pk") { + sql(s""" + |CREATE TABLE $DEFAULT_DATABASE.t_lake_tr_pk (id INT, name STRING, score INT) + | TBLPROPERTIES ( + | '${ConfigOptions.TABLE_DATALAKE_ENABLED.key()}' = true, + | '${ConfigOptions.TABLE_DATALAKE_FRESHNESS.key()}' = '1s', + | '${PRIMARY_KEY.key()}' = 'id', + | '${BUCKET_NUMBER.key()}' = 1) + |""".stripMargin) + + // group 1 (before the window) -> tiered to lake, still retained in Fluss changelog + sql(s""" + |INSERT INTO $DEFAULT_DATABASE.t_lake_tr_pk VALUES + |(1, "alice", 90), (2, "bob", 85), (3, "charlie", 95) + |""".stripMargin) + tierToLake("t_lake_tr_pk") + + val t1 = System.currentTimeMillis() + Thread.sleep(50) + // group 2 (inside the window): update id=2, insert id=4 + sql(s""" + |INSERT INTO $DEFAULT_DATABASE.t_lake_tr_pk VALUES + |(2, "bob_updated", 100), (4, "david", 88) + |""".stripMargin) + Thread.sleep(500) + val t2 = System.currentTimeMillis() + Thread.sleep(50) + // group 3 (after the window): update id=1, insert id=5 + sql(s""" + |INSERT INTO $DEFAULT_DATABASE.t_lake_tr_pk VALUES + |(1, "alice_updated", 91), (5, "eve", 92) + |""".stripMargin) + Thread.sleep(200) + + val df = + sql(s"SELECT * FROM $TVF('$DEFAULT_DATABASE.t_lake_tr_pk', '$t1', '$t2') ORDER BY id") + val partitions = lakeInputPartitions(df) + assert(partitions.nonEmpty, "expected at least one Fluss changelog partition") + assert( + partitions.forall { + case p: FlussUpsertInputPartition => p.snapshotId == -1 + case _ => false + }, + s"time-range read must be log-only with no kv/lake snapshot (snapshotId == -1), " + + s"got: ${partitions.mkString(", ")}" + ) + // Only keys inserted/updated within [t1, t2): id=2 (updated), id=4 (inserted). + checkAnswer(df, Row(2, "bob_updated", 100) :: Row(4, "david", 88) :: Nil) + } + } +} + +@SparkLakeTest +class SparkLakePaimonTimeRangeReadTest extends SparkLakeTimeRangeReadTest { + + override protected def dataLakeFormat: DataLakeFormat = DataLakeFormat.PAIMON + + override protected def flussConf: Configuration = { + val conf = super.flussConf + conf.setString("datalake.format", DataLakeFormat.PAIMON.toString) + conf.setString("datalake.paimon.metastore", "filesystem") + conf.setString("datalake.paimon.cache-enabled", "false") + warehousePath = + Files.createTempDirectory("fluss-testing-paimon-timerange-lake").resolve("warehouse").toString + conf.setString("datalake.paimon.warehouse", warehousePath) + conf + } + + override protected def lakeCatalogConf: Configuration = { + val conf = new Configuration() + conf.setString("metastore", "filesystem") + conf.setString("warehouse", warehousePath) + conf + } +} diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/read/FlussOffsetInitializersTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/read/FlussOffsetInitializersTest.scala new file mode 100644 index 0000000000..9e24fcdff2 --- /dev/null +++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/read/FlussOffsetInitializersTest.scala @@ -0,0 +1,100 @@ +/* + * 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.fluss.spark.read + +import org.apache.fluss.config.Configuration +import org.apache.fluss.spark.SparkFlussConf + +import org.apache.spark.sql.util.CaseInsensitiveStringMap +import org.assertj.core.api.Assertions.assertThat +import org.scalatest.funsuite.AnyFunSuite + +/** + * Unit tests for how scan options are resolved into offset initializers, and for the retention + * guard of an incremental (time-range) read. The end-to-end behavior is covered by + * [[org.apache.fluss.spark.SparkTimeRangeTvfTest]]. + */ +class FlussOffsetInitializersTest extends AnyFunSuite { + + private def scanOptions(entries: (String, String)*): CaseInsensitiveStringMap = { + val map = new java.util.HashMap[String, String]() + entries.foreach { case (k, v) => map.put(k, v) } + new CaseInsensitiveStringMap(map) + } + + test("incremental read is enabled by the presence of a start timestamp") { + val startKey = SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP.key() + assertThat(FlussOffsetInitializers.isIncrementalRead(scanOptions())).isFalse + assertThat(FlussOffsetInitializers.isIncrementalRead(scanOptions(startKey -> " "))).isFalse + assertThat( + FlussOffsetInitializers.isIncrementalRead(scanOptions(startKey -> "1767225600000"))).isTrue + } + + test("scan.incremental.timestamp.out-of-range toggles fail-fast (default error)") { + val key = SparkFlussConf.SCAN_INCREMENTAL_TIMESTAMP_OUT_OF_RANGE.key() + // default (unset) is error -> fail fast + assertThat(FlussOffsetInitializers.failOnTimestampOutOfRange(scanOptions())).isTrue + assertThat( + FlussOffsetInitializers.failOnTimestampOutOfRange(scanOptions(key -> "error"))).isTrue + assertThat( + FlussOffsetInitializers.failOnTimestampOutOfRange(scanOptions(key -> "ERROR"))).isTrue + // adjust -> clamp instead of failing + assertThat( + FlussOffsetInitializers.failOnTimestampOutOfRange(scanOptions(key -> "adjust"))).isFalse + } + + test("retention guard decision (isBeforeRetention)") { + // brand-new bucket (earliest == 0) is never flagged, even for a very old start offset + assertThat(FlussOffsetInitializers.isBeforeRetention(0L, 0L)).isFalse + assertThat(FlussOffsetInitializers.isBeforeRetention(5L, 0L)).isFalse + // a trimmed bucket (earliest > 0) is flagged when the start lands at or before earliest + assertThat(FlussOffsetInitializers.isBeforeRetention(10L, 10L)).isTrue + assertThat(FlussOffsetInitializers.isBeforeRetention(3L, 10L)).isTrue + // a start strictly after earliest is within retention + assertThat(FlussOffsetInitializers.isBeforeRetention(11L, 10L)).isFalse + } + + test("TTL-exceeded start fails fast with a table.log.ttl hint") { + // start at/before a trimmed earliest (earliest > 0): fail fast with a clear TTL message + val ex = intercept[IllegalArgumentException] { + FlussOffsetInitializers.requireStartWithinRetention("fluss.t", "dt=2026", 2, 5L, 10L) + } + assertThat(ex.getMessage).contains("table.log.ttl") + assertThat(ex.getMessage).contains("bucket 2") + assertThat(ex.getMessage).contains("partition 'dt=2026'") + } + + test("invalid start timestamp format fails with the option name") { + val startKey = SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP.key() + val ex = intercept[IllegalArgumentException] { + FlussOffsetInitializers.incrementalStartOffsetsInitializer( + scanOptions(startKey -> "not-a-timestamp")) + } + assertThat(ex.getMessage).contains(startKey) + } + + test("scan.startup.mode=timestamp is not a batch option") { + val ex = intercept[IllegalArgumentException] { + FlussOffsetInitializers.startOffsetsInitializer( + scanOptions(SparkFlussConf.SCAN_START_UP_MODE.key() -> "timestamp"), + new Configuration()) + } + assertThat(ex.getMessage).contains("Unsupported scan start up mode") + assertThat(ex.getMessage).contains(SparkFlussConf.SCAN_INCREMENTAL_START_TIMESTAMP.key()) + } +} diff --git a/website/docs/engine-spark/options.md b/website/docs/engine-spark/options.md index b1f74cbc5e..784ba8ec1d 100644 --- a/website/docs/engine-spark/options.md +++ b/website/docs/engine-spark/options.md @@ -14,6 +14,16 @@ The following Spark configurations can be used to control read behavior for both | Option | Default | Description | |--------|---------|-------------| -| `spark.sql.fluss.scan.startup.mode` | `full` | The startup mode when reading a Fluss table. Supported values: **Note:** For Structured Streaming read, only `latest` mode is currently supported. | +| `spark.sql.fluss.scan.startup.mode` | `full` | The startup mode when reading a Fluss table. Supported values: **Note:** This option only affects Structured Streaming reads, and only `latest` mode is currently supported there. Batch reads ignore it: a plain batch read is always the full table, and a time-range batch read is requested per query (see below). | | `spark.sql.fluss.read.optimized` | `false` | If `true`, Spark will only read data from the data lake snapshot or KV snapshot, without merging log changes. This can improve read performance but may return stale data for primary key tables. | | `spark.sql.fluss.scan.poll.timeout` | `10000ms` | The timeout for the log scanner to poll records. | + +## Per-Query Read Options + +The following options configure a single read and are **not** read from session configuration, so a time window can never leak into later reads. In SQL they are set by the `fluss_incremental_between_timestamp(...)` table-valued function; in the DataFrame API by `spark.read.option(...)` (without the `spark.sql.fluss.` prefix). See [Reads](reads.md#time-range-batch-read) for the full semantics. + +| Option | Default | Description | +|--------|---------|-------------| +| `scan.incremental.start.timestamp` | (none) | Enables an incremental (time-range) batch read and sets the **inclusive** lower bound of the window. Accepts epoch milliseconds (e.g. `1678883047356`) or a `yyyy-MM-dd HH:mm:ss` datetime (e.g. `2023-12-09 23:09:12`) interpreted in the Spark session time zone (`spark.sql.session.timeZone`). Batch read only; it has no effect on streaming reads. If the timestamp predates the data still retained by Fluss (bounded by `table.log.ttl`), behavior is controlled by `scan.incremental.timestamp.out-of-range`. | +| `scan.incremental.end.timestamp` | `latest` | The **exclusive** upper bound of an incremental batch read, producing a left-closed right-open `[start, end)` window. `latest` (default) stops at the latest committed data captured at planning time; otherwise the same value format as `scan.incremental.start.timestamp`. Only honored when `scan.incremental.start.timestamp` is set. A timestamp in the future is rejected by the server (`InvalidTimestampException`). | +| `scan.incremental.timestamp.out-of-range` | `error` | Behavior when `scan.incremental.start.timestamp` precedes the earliest data still retained by Fluss (bounded by `table.log.ttl`). | diff --git a/website/docs/engine-spark/reads.md b/website/docs/engine-spark/reads.md index ca48bdb201..5335bceb32 100644 --- a/website/docs/engine-spark/reads.md +++ b/website/docs/engine-spark/reads.md @@ -254,6 +254,86 @@ INSERT INTO fluss_order_with_lake VALUES SELECT SUM(total_price) AS total_revenue FROM fluss_order_with_lake; ``` +## Time-Range Batch Read + +A time-range batch read returns the data written to Fluss within a `[start, end)` time window (left-closed, right-open on the commit timestamp). This is the building block for incremental pipelines, e.g. an hourly job that reads only the rows written in the past hour. + +The timestamp value is either epoch milliseconds or a `yyyy-MM-dd HH:mm:ss` datetime interpreted in the Spark session time zone (`spark.sql.session.timeZone`). + +### Output semantics + +| Table type | Output of a `[t1, t2)` read | +|------------|-----------------------------| +| **Log table** | Every record appended within the window. | +| **Primary key table** | The rows whose keys were **inserted or updated** within the window, folded to their latest value as of `t2`. Keys that were only deleted in the window are excluded (i.e. the result keeps `+I`/`+U` after-images and drops `-U`/`-D`). | +| **Lake-enabled table** | Same as the underlying log or primary key table, but read **only from Fluss**. A time-range read never unions the lake snapshot, so it is limited to the data still retained in Fluss (see below). | + +### Using the table-valued function (recommended for SQL) + +`fluss_incremental_between_timestamp(table, start[, end])` reads a time window in a single statement. Omit `end` to read up to the latest committed data at planning time. + +```sql title="Spark SQL" +-- Read the past hour on a log table (epoch-millis form) +SELECT * FROM fluss_incremental_between_timestamp('log_table', '1767225600000', '1767312000000') +ORDER BY order_id; +``` + +```sql title="Spark SQL" +-- Incremental changes on a primary key table (datetime form) +-- Returns the latest value of every key changed in the window; deleted keys are excluded +SELECT * FROM fluss_incremental_between_timestamp( + 'pk_table', '2026-01-01 00:00:00', '2026-01-02 00:00:00') +ORDER BY order_id; +``` + +```sql title="Spark SQL" +-- Omit the end to read from a start timestamp up to the latest data +SELECT * FROM fluss_incremental_between_timestamp('log_table', '2026-01-01 00:00:00'); +``` + +The start/end arguments may be any constant expression, so a rolling window does not need to be computed outside SQL: + +```sql title="Spark SQL" +-- The past hour, as epoch milliseconds (unix_timestamp() returns seconds) +SELECT * FROM fluss_incremental_between_timestamp( + 'log_table', + CAST((unix_timestamp() - 3600) * 1000 AS STRING), + CAST(unix_timestamp() * 1000 AS STRING)); + +-- The past hour, as datetime strings +SELECT * FROM fluss_incremental_between_timestamp( + 'log_table', + date_format(now() - INTERVAL 1 HOUR, 'yyyy-MM-dd HH:mm:ss'), + date_format(now(), 'yyyy-MM-dd HH:mm:ss')); +``` + +The table argument is a string and accepts `table`, `database.table` or `catalog.database.table`; unqualified names resolve against the current catalog and database. The start/end arguments accept a string (epoch milliseconds or `yyyy-MM-dd HH:mm:ss`), an integral epoch-milliseconds value, or a `TIMESTAMP`, and may be produced by constant expressions such as the datetime functions above (column references are not allowed). The result is an ordinary relation, so projection, filters and joins work as usual. + +:::note +The function is provided by the Fluss Spark session extension, so `spark.sql.extensions=org.apache.fluss.spark.FlussSparkSessionExtensions` must be configured (see [Getting Started](getting-started.md)). Its options apply to that single query only. + +Fluss uses `[start, end)` (start inclusive, end exclusive). This differs from Paimon's similarly named `paimon_incremental_between_timestamp`, which is start-exclusive and end-inclusive. +::: + +### Using the DataFrame API + +The same window can be expressed with scan options, which is how it is configured from the DataFrame API. Setting `scan.incremental.start.timestamp` is what turns a batch read into an incremental one; `scan.incremental.end.timestamp` defaults to the reserved value `latest`: + +```scala title="Spark Scala" +spark.read + .option("scan.incremental.start.timestamp", "2026-01-01 00:00:00") + .option("scan.incremental.end.timestamp", "2026-01-02 00:00:00") + .table("fluss_catalog.fluss.log_table") +``` + +:::note +The `scan.incremental.*` options are per-query read options only. Unlike the options in [Options](options.md), they are **not** picked up from session configuration (`SET spark.sql.fluss.scan.incremental...` has no effect), which keeps a time window from silently applying to later reads in the same session. +::: + +:::warning Retention boundary +A time-range read only sees data still retained by Fluss, which is bounded by `table.log.ttl` (default 7 days). If the start timestamp predates the earliest retained data, the default behavior (`scan.incremental.timestamp.out-of-range=error`) **fails fast** with a clear error instead of silently returning a truncated window — narrow the time range or increase `table.log.ttl`. Set `scan.incremental.timestamp.out-of-range=adjust` to instead clamp the start to the earliest retained data and read from there. An end timestamp in the future is rejected by the server. Reading data older than the Fluss retention (including from tiered lake storage) is not supported by this mode. +::: + ## All Data Types Fluss Spark connector supports reading all Fluss data types including nested types: diff --git a/website/docs/engine-spark/structured-streaming.md b/website/docs/engine-spark/structured-streaming.md index 4fcf2fafd7..0f6e3ba9ac 100644 --- a/website/docs/engine-spark/structured-streaming.md +++ b/website/docs/engine-spark/structured-streaming.md @@ -51,7 +51,7 @@ Fluss supports exactly-once semantics for streaming writes through Spark's check Fluss supports reading data from Fluss tables using Spark Structured Streaming. The streaming source continuously reads new data as it arrives. :::caution Limitations -- Streaming read currently only supports the `latest` startup mode. Other modes (`full`, `earliest`, `timestamp`) are not yet supported and will be available in a future release. +- Streaming read currently only supports the `latest` startup mode. Other modes (`full`, `earliest`) are not yet supported and will be available in a future release. To read a bounded time range, use an incremental batch read instead (see [Reads](reads.md#time-range-batch-read)). ::: ### Read from Log Table