You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/09/24 16:42:47 UTC
(spark) branch branch-4.x updated: [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 33fe09d94fe0 [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes
33fe09d94fe0 is described below
commit 33fe09d94fe039a8ec9099d0b78ebe6933ac1128
Author: Joel Robin P <jo...@gmail.com>
AuthorDate: Thu Sep 24 16:42:16 2026 +0000
[SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes
### What changes were proposed in this pull request?
This PR adds `FileIndex.isTimeTravel`, defaulting to `false`, so V1 file data sources can
identify immutable time-travel snapshots.
`CacheManager` honors this signal during write-driven cache invalidation by table name and path.
`InsertIntoHadoopFsRelationCommand` marks its path-based recache as write-driven. Existing explicit
refresh operations continue to include time-travel snapshots.
### Why are the changes needed?
[SPARK-53732](https://issues.apache.org/jira/browse/SPARK-53732) changed write-driven cache
invalidation from plan matching to table-name matching with `includeTimeTravel = false`. V2
relations honor that flag through `DataSourceV2Relation.timeTravelSpec`, but V1
`LogicalRelationWithTable` was matched only by table name.
For providers that lower a pinned read to `LogicalRelation(HadoopFsRelation(FileIndex))`, the
time-travel state survives only in the `FileIndex`. A write to the live table therefore cleared and
rebuilt the pinned cache. The DataFrame remained registered as cached, but the next action
re-executed its plan, including expensive or non-idempotent UDFs.
The new signal closes this V1 gap without coupling Spark to a specific data source.
JIRA: [SPARK-59631](https://issues.apache.org/jira/browse/SPARK-59631)
### Does this PR introduce _any_ user-facing change?
Yes. For V1 data sources that identify an immutable snapshot, appending to or overwriting the live
table no longer recomputes an already materialized cache for that snapshot. Live-table caches are
still refreshed after writes, and explicit refresh or uncache operations still include pinned
snapshots.
Before:
```text
is_cached=true
udf_calls=10->20
tokens_same=false
```
After the coordinated Spark and provider changes:
```text
is_cached=true
udf_calls=10->10
tokens_same=true
```
### How was this patch tested?
Added ten tests in `V1TimeTravelCacheSuite` covering name- and path-based invalidation for pinned
and live V1 relations, including a mixed pinned/live union, a real V1 Parquet append through `InsertIntoHadoopFsRelationCommand`, and V2
path invalidation for both time-travel and live `ExtractV2Table` relations. The production-write
test verifies that the pinned cache remains materialized while the live cache is cleared and
refreshed. The V2 tests verify that write-driven invalidation preserves a pinned relation but still
invalidates a live relation, while explicit refresh includes the pinned relation.
`V1TimeTravelCacheSuite`, the focused `CachedTableSuite` time-travel regression, SQL scalastyle,
test scalastyle, and MiMa checks passed.
End-to-end testing reproduced `10->20` UDF executions with unpatched Spark and preserved `10->10`
with the coordinated Spark/provider changes. The provider integration matrix passed all four tests
in both `AUTO` and `NONE` modes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (gpt-5.6-sol)
Closes #58907 from joelrobin18/fix-v1-time-travel-cache-recache.
Authored-by: Joel Robin P <jo...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 269bae36567e9d8f8727b1e6b6c585c5f69909ad)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../apache/spark/sql/execution/CacheManager.scala | 69 ++++-
.../sql/execution/datasources/FileIndex.scala | 6 +
.../InsertIntoHadoopFsRelationCommand.scala | 6 +-
.../apache/spark/sql/V1TimeTravelCacheSuite.scala | 335 +++++++++++++++++++++
4 files changed, 400 insertions(+), 16 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
index 01fedf51602f..b3bf3b8ab46d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
@@ -39,9 +39,14 @@ import org.apache.spark.sql.connector.catalog.transactions.Transaction
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.columnar.InMemoryRelation
import org.apache.spark.sql.execution.command.CommandUtils
-import org.apache.spark.sql.execution.datasources.{FileIndex, HadoopFsRelation, LogicalRelation, LogicalRelationWithTable}
+import org.apache.spark.sql.execution.datasources.{
+ FileIndex,
+ HadoopFsRelation,
+ LogicalRelation,
+ LogicalRelationWithTable}
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2Relation, ExtractV2CatalogAndIdentifier, ExtractV2Table, FileTable, V2TableRefreshUtil}
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK
@@ -255,8 +260,9 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
includeTimeTravel: Boolean): Boolean = {
EliminateSubqueryAliases(plan) match {
- case LogicalRelationWithTable(_, Some(catalogTable)) =>
- isSameName(name, catalogTable.identifier.nameParts, resolver)
+ case LogicalRelationWithTable(relation, Some(catalogTable)) =>
+ isSameName(name, catalogTable.identifier.nameParts, resolver) &&
+ (includeTimeTravel || !isTimeTravelRelation(relation))
case DataSourceV2Relation(_, _, Some(catalog), Some(v2Ident), _, timeTravelSpec) =>
val nameInCache = v2Ident.toQualifiedNameParts(catalog)
@@ -272,6 +278,11 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
}
}
+ private def isTimeTravelRelation(relation: BaseRelation): Boolean = relation match {
+ case hadoopFsRelation: HadoopFsRelation => hadoopFsRelation.location.isTimeTravel
+ case _ => false
+ }
+
private def isSameName(
name: Seq[String],
catalog: CatalogPlugin,
@@ -582,8 +593,24 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* `HadoopFsRelation` node(s) as part of its logical plan.
*/
def recacheByPath(spark: SparkSession, resourcePath: Path, fs: FileSystem): Unit = {
+ recacheByPath(spark, resourcePath, fs, includeTimeTravel = true)
+ }
+
+ /**
+ * Tries to re-cache matching entries whose logical plans contain a file-backed relation under
+ * `resourcePath`. If `includeTimeTravel` is false, immutable time-travel relations are ignored
+ * when deciding whether an entry must be re-cached; another matching live relation in the same
+ * plan can still trigger re-caching.
+ */
+ def recacheByPath(
+ spark: SparkSession,
+ resourcePath: Path,
+ fs: FileSystem,
+ includeTimeTravel: Boolean): Unit = {
val qualifiedPath = fs.makeQualified(resourcePath)
- recacheByCondition(spark, _.plan.exists(lookupAndRefresh(_, fs, qualifiedPath)))
+ recacheByCondition(
+ spark,
+ _.plan.exists(lookupAndRefresh(_, fs, qualifiedPath, includeTimeTravel)))
}
/**
@@ -592,33 +619,45 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* in the plan. If found, we refresh the metadata and return true. Otherwise, this method returns
* false.
*/
- private def lookupAndRefresh(plan: LogicalPlan, fs: FileSystem, qualifiedPath: Path): Boolean = {
+ private def lookupAndRefresh(
+ plan: LogicalPlan,
+ fs: FileSystem,
+ qualifiedPath: Path,
+ includeTimeTravel: Boolean): Boolean = {
plan match {
case lr: LogicalRelation => lr.relation match {
case hr: HadoopFsRelation =>
- refreshFileIndexIfNecessary(hr.location, fs, qualifiedPath)
+ refreshFileIndexIfNecessary(hr.location, fs, qualifiedPath, includeTimeTravel)
case _ => false
}
- case ExtractV2Table(fileTable: FileTable) =>
- refreshFileIndexIfNecessary(fileTable.fileIndex, fs, qualifiedPath)
+ case relation @ ExtractV2Table(fileTable: FileTable)
+ if includeTimeTravel || relation.timeTravelSpec.isEmpty =>
+ refreshFileIndexIfNecessary(
+ fileTable.fileIndex,
+ fs,
+ qualifiedPath,
+ includeTimeTravel)
case _ => false
}
}
/**
- * Refresh the given [[FileIndex]] if any of its root paths is a subdirectory
- * of the `qualifiedPath`.
- * @return whether the [[FileIndex]] is refreshed.
+ * Refresh the given [[FileIndex]] if one of its root paths is a subdirectory of
+ * `qualifiedPath` and either `includeTimeTravel` is true or the index is not an immutable
+ * time-travel snapshot.
+ * @return whether the [[FileIndex]] was refreshed.
*/
private def refreshFileIndexIfNecessary(
fileIndex: FileIndex,
fs: FileSystem,
- qualifiedPath: Path): Boolean = {
- val needToRefresh = fileIndex.rootPaths
- .map(_.makeQualified(fs.getUri, fs.getWorkingDirectory))
- .exists(isSubDir(qualifiedPath, _))
+ qualifiedPath: Path,
+ includeTimeTravel: Boolean): Boolean = {
+ val needToRefresh = (includeTimeTravel || !fileIndex.isTimeTravel) &&
+ fileIndex.rootPaths
+ .map(_.makeQualified(fs.getUri, fs.getWorkingDirectory))
+ .exists(isSubDir(qualifiedPath, _))
if (needToRefresh) fileIndex.refresh()
needToRefresh
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala
index 0291a5fd28a7..b4be11041820 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala
@@ -149,6 +149,12 @@ trait FileIndex {
/** Schema of the partitioning columns, or the empty schema if the table is not partitioned. */
def partitionSchema: StructType
+ /**
+ * Returns whether this index reads an immutable time-travel snapshot. Table writes do not need
+ * to invalidate cached snapshots whose contents cannot change.
+ */
+ def isTimeTravel: Boolean = false
+
/**
* Returns an optional metadata operation time, in nanoseconds, for listing files.
*
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
index d969dbcd8ea9..4c59ad2933bb 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
@@ -210,7 +210,11 @@ case class InsertIntoHadoopFsRelationCommand(
// refresh cached files in FileIndex
fileIndex.foreach(_.refresh())
// refresh data cache if table is cached
- sparkSession.sharedState.cacheManager.recacheByPath(sparkSession, outputPath, fs)
+ sparkSession.sharedState.cacheManager.recacheByPath(
+ sparkSession,
+ outputPath,
+ fs,
+ includeTimeTravel = false)
if (catalogTable.nonEmpty) {
CommandUtils.updateTableStats(sparkSession, catalogTable.get)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/V1TimeTravelCacheSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/V1TimeTravelCacheSuite.scala
new file mode 100644
index 000000000000..57d0c80f18e4
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/V1TimeTravelCacheSuite.scala
@@ -0,0 +1,335 @@
+/*
+ * 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
+
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.AsOfVersion
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.classic.{Dataset => ClassicDataset}
+import org.apache.spark.sql.execution.datasources.{
+ FileIndex,
+ HadoopFsRelation,
+ LogicalRelation,
+ PartitionDirectory}
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetTable
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.sources.BaseRelation
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+class V1TimeTravelCacheSuite extends QueryTest with SharedSparkSession {
+
+ private val cachedTableNameParts = Seq("db", "tbl")
+ private val tableSchema = StructType(Seq(StructField("id", IntegerType, nullable = false)))
+ private val tableMetadata = CatalogTable(
+ identifier = TableIdentifier("tbl", Some("db")),
+ tableType = CatalogTableType.MANAGED,
+ storage = CatalogStorageFormat.empty,
+ schema = tableSchema,
+ provider = Some("test"))
+
+ override def afterEach(): Unit = {
+ try {
+ spark.catalog.clearCache()
+ } finally {
+ super.afterEach()
+ }
+ }
+
+ test("recacheTableOrView excludes a time travel V1 FileIndex") {
+ assertRecacheBehavior(
+ newFileRelation(isTimeTravel = true),
+ includeTimeTravel = false,
+ expectedToRecache = false)
+ }
+
+ test("recacheTableOrView includes a live V1 FileIndex") {
+ assertRecacheBehavior(
+ newFileRelation(isTimeTravel = false),
+ includeTimeTravel = false,
+ expectedToRecache = true)
+ }
+
+ test("recacheTableOrView includes a time travel V1 FileIndex when requested") {
+ assertRecacheBehavior(
+ newFileRelation(isTimeTravel = true),
+ includeTimeTravel = true,
+ expectedToRecache = true)
+ }
+
+ test("write-driven recacheByPath excludes a time travel V1 FileIndex") {
+ assertRecacheByPathBehavior(
+ isTimeTravel = true,
+ writeDriven = true,
+ expectedToRecache = false)
+ }
+
+ test("write-driven recacheByPath includes a live V1 FileIndex") {
+ assertRecacheByPathBehavior(
+ isTimeTravel = false,
+ writeDriven = true,
+ expectedToRecache = true)
+ }
+
+ test("explicit recacheByPath includes a time travel V1 FileIndex") {
+ assertRecacheByPathBehavior(
+ isTimeTravel = true,
+ writeDriven = false,
+ expectedToRecache = true)
+ }
+
+ test("write-driven recacheByPath rebuilds a mixed pinned and live V1 cache") {
+ withTempDir { dir =>
+ val rootPath = new Path(dir.toURI)
+ val pinnedIndex = new TestFileIndex(isTimeTravel = true, rootPaths = Seq(rootPath))
+ val liveIndex = new TestFileIndex(isTimeTravel = false, rootPaths = Seq(rootPath))
+ val pinned = ClassicDataset.ofRows(
+ spark,
+ LogicalRelation(newFileRelation(pinnedIndex), tableMetadata))
+ val live = ClassicDataset.ofRows(
+ spark,
+ LogicalRelation(newFileRelation(liveIndex), tableMetadata))
+ // Keep the pinned child first to ensure traversal continues past it to the live child.
+ val compound = pinned.union(live).persist()
+ try {
+ compound.count()
+ assertCacheLoading(compound, expected = true)
+
+ val fs = rootPath.getFileSystem(spark.sessionState.newHadoopConf())
+ spark.sharedState.cacheManager.recacheByPath(
+ spark,
+ rootPath,
+ fs,
+ includeTimeTravel = false)
+
+ assertCacheLoading(compound, expected = false)
+ assert(pinnedIndex.refreshCount === 0)
+ assert(liveIndex.refreshCount === 1)
+ } finally {
+ compound.unpersist(blocking = true)
+ }
+ }
+ }
+
+ test("V1 file write preserves a time travel cache and refreshes a live cache") {
+ withTempPath { path =>
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "parquet") {
+ val dataPath = path.getCanonicalPath
+ spark.range(1).write.parquet(dataPath)
+
+ val pinnedIndex = new TestFileIndex(
+ isTimeTravel = true,
+ rootPaths = Seq(new Path(path.toURI)))
+ val pinned = ClassicDataset.ofRows(
+ spark,
+ LogicalRelation(newFileRelation(pinnedIndex), tableMetadata)).persist()
+ val live = spark.read.parquet(dataPath).persist()
+ try {
+ pinned.count()
+ checkAnswer(live, Row(0L))
+ assertCacheLoading(pinned, expected = true)
+ assertCacheLoading(live, expected = true)
+
+ spark.range(1, 2).write.mode("append").parquet(dataPath)
+
+ assertCacheLoading(pinned, expected = true)
+ assertCacheLoading(live, expected = false)
+ assert(pinnedIndex.refreshCount === 0)
+ checkAnswer(spark.read.parquet(dataPath), Seq(Row(0L), Row(1L)))
+ assertCacheLoading(live, expected = true)
+ } finally {
+ pinned.unpersist(blocking = true)
+ live.unpersist(blocking = true)
+ }
+ }
+ }
+ }
+
+ test("recacheByPath excludes a V2 time-travel relation only for write-driven refresh") {
+ withTempDir { dir =>
+ val rootPath = new Path(dir.toURI)
+ val options = CaseInsensitiveStringMap.empty()
+ val table = ParquetTable(
+ name = "time-travel-table",
+ sparkSession = spark,
+ options = options,
+ paths = Seq(rootPath.toString),
+ userSpecifiedSchema = Some(tableSchema),
+ fallbackFileFormat = classOf[ParquetFileFormat])
+ assert(!table.fileIndex.isTimeTravel)
+
+ val relation = DataSourceV2Relation.create(
+ table,
+ catalog = None,
+ identifier = None,
+ options = options,
+ timeTravelSpec = Some(AsOfVersion("v1")))
+ val df = ClassicDataset.ofRows(spark, relation).persist()
+ try {
+ df.count()
+ assertCacheLoading(df, expected = true)
+
+ val resourcePath = table.fileIndex.rootPaths.head
+ val fs = resourcePath.getFileSystem(spark.sessionState.newHadoopConf())
+ spark.sharedState.cacheManager.recacheByPath(
+ spark,
+ resourcePath,
+ fs,
+ includeTimeTravel = false)
+ assertCacheLoading(df, expected = true)
+
+ spark.sharedState.cacheManager.recacheByPath(spark, resourcePath, fs)
+ assertCacheLoading(df, expected = false)
+ } finally {
+ df.unpersist(blocking = true)
+ }
+ }
+ }
+
+ test("write-driven recacheByPath includes a live V2 relation") {
+ withTempDir { dir =>
+ val rootPath = new Path(dir.toURI)
+ val options = CaseInsensitiveStringMap.empty()
+ val table = ParquetTable(
+ name = "live-table",
+ sparkSession = spark,
+ options = options,
+ paths = Seq(rootPath.toString),
+ userSpecifiedSchema = Some(tableSchema),
+ fallbackFileFormat = classOf[ParquetFileFormat])
+ assert(!table.fileIndex.isTimeTravel)
+
+ val relation = DataSourceV2Relation.create(
+ table,
+ catalog = None,
+ identifier = None,
+ options = options,
+ timeTravelSpec = None)
+ val df = ClassicDataset.ofRows(spark, relation).persist()
+ try {
+ df.count()
+ assertCacheLoading(df, expected = true)
+
+ val resourcePath = table.fileIndex.rootPaths.head
+ val fs = resourcePath.getFileSystem(spark.sessionState.newHadoopConf())
+ spark.sharedState.cacheManager.recacheByPath(
+ spark,
+ resourcePath,
+ fs,
+ includeTimeTravel = false)
+ assertCacheLoading(df, expected = false)
+ } finally {
+ df.unpersist(blocking = true)
+ }
+ }
+ }
+
+ private def assertRecacheBehavior(
+ relation: BaseRelation,
+ includeTimeTravel: Boolean,
+ expectedToRecache: Boolean): Unit = {
+ val df = ClassicDataset.ofRows(spark, LogicalRelation(relation, tableMetadata)).persist()
+ try {
+ df.count()
+ assertCacheLoading(df, expected = true)
+
+ spark.sharedState.cacheManager.recacheTableOrView(
+ spark,
+ cachedTableNameParts,
+ includeTimeTravel = includeTimeTravel)
+
+ assertCacheLoading(df, expected = !expectedToRecache)
+ } finally {
+ df.unpersist(blocking = true)
+ }
+ }
+
+ private def assertCacheLoading(df: ClassicDataset[_], expected: Boolean): Unit = {
+ val cachedData = spark.sharedState.cacheManager
+ .lookupCachedData(df)
+ .getOrElse(fail("DataFrame is not cached"))
+ assert(cachedData.cachedRepresentation.cacheBuilder.isCachedColumnBuffersLoaded === expected)
+ }
+
+ private def assertRecacheByPathBehavior(
+ isTimeTravel: Boolean,
+ writeDriven: Boolean,
+ expectedToRecache: Boolean): Unit = {
+ withTempDir { dir =>
+ val rootPath = new Path(dir.toURI)
+ val fileIndex = new TestFileIndex(isTimeTravel, Seq(rootPath))
+ val df = ClassicDataset.ofRows(
+ spark,
+ LogicalRelation(newFileRelation(fileIndex), tableMetadata)).persist()
+ try {
+ df.count()
+ assertCacheLoading(df, expected = true)
+
+ val fs = rootPath.getFileSystem(spark.sessionState.newHadoopConf())
+ if (writeDriven) {
+ spark.sharedState.cacheManager.recacheByPath(
+ spark,
+ rootPath,
+ fs,
+ includeTimeTravel = false)
+ } else {
+ spark.sharedState.cacheManager.recacheByPath(spark, rootPath, fs)
+ }
+
+ assertCacheLoading(df, expected = !expectedToRecache)
+ assert(fileIndex.refreshCount === (if (expectedToRecache) 1 else 0))
+ } finally {
+ df.unpersist(blocking = true)
+ }
+ }
+ }
+
+ private def newFileRelation(isTimeTravel: Boolean): HadoopFsRelation = {
+ newFileRelation(new TestFileIndex(isTimeTravel))
+ }
+
+ private def newFileRelation(fileIndex: FileIndex): HadoopFsRelation = {
+ HadoopFsRelation(
+ location = fileIndex,
+ partitionSchema = StructType(Nil),
+ dataSchema = tableSchema,
+ bucketSpec = None,
+ fileFormat = new ParquetFileFormat,
+ options = Map.empty)(spark)
+ }
+
+ private class TestFileIndex(
+ override val isTimeTravel: Boolean,
+ override val rootPaths: Seq[Path] = Nil) extends FileIndex {
+ var refreshCount: Int = 0
+
+ override def listFiles(
+ partitionFilters: Seq[Expression],
+ dataFilters: Seq[Expression]): Seq[PartitionDirectory] = Nil
+ override def inputFiles: Array[String] = Array.empty
+ override def refresh(): Unit = refreshCount += 1
+ override def sizeInBytes: Long = 0L
+ override def partitionSchema: StructType = StructType(Nil)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org