JingsongLi commented on code in PR #8300:
URL: https://github.com/apache/paimon/pull/8300#discussion_r3446430724
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/SparkFormatTable.scala:
##########
@@ -107,28 +116,231 @@ object SparkFormatTable {
inferred
}
}
+}
- // Extend from InMemoryFileIndex to override partitionSchema
- private class PartitionedInMemoryFileIndex(
+/**
+ * A lazy [[PartitioningAwareFileIndex]] that defers file listing until
[[listFiles]] is called, and
+ * prunes partition directories level-by-level using partition filters.
Inspired by Spark's
+ * [[CatalogFileIndex]] which queries the metastore before listing; here we
discover matching
+ * partitions from the filesystem with per-level filter evaluation.
+ */
+class LazyPartitionPruningFileIndex(
+ sparkSession: SparkSession,
+ tableLocations: Seq[Path],
+ parameters: Map[String, String],
+ userSpecifiedSchema: Option[StructType],
+ fileStatusCache: FileStatusCache,
+ _partitionSchema: StructType)
+ extends PartitioningAwareFileIndex(sparkSession, parameters,
userSpecifiedSchema, fileStatusCache)
+ with Logging {
+
+ override val rootPaths: Seq[Path] = tableLocations
+
+ override def partitionSchema: StructType = _partitionSchema
+
+ private lazy val fullIndex: FullListingFileIndex = {
+ new FullListingFileIndex(
+ sparkSession,
+ tableLocations,
+ parameters,
+ userSpecifiedSchema,
+ fileStatusCache)
+ }
+
+ override protected def leafFiles
+ : mutable.LinkedHashMap[Path, _root_.org.apache.hadoop.fs.FileStatus] =
+ fullIndex.exposedLeafFiles
+
+ override protected def leafDirToChildrenFiles
+ : Map[Path, Array[_root_.org.apache.hadoop.fs.FileStatus]] =
+ fullIndex.exposedLeafDirToChildrenFiles
+
+ override def partitionSpec(): PartitionSpec =
+ SparkFormatTable.alignPartitionSpec(fullIndex.partitionSpec(),
_partitionSchema)
+
+ private class FullListingFileIndex(
sparkSession: SparkSession,
- rootPathsSpecified: Seq[Path],
+ rootPaths: Seq[Path],
parameters: Map[String, String],
userSpecifiedSchema: Option[StructType],
- fileStatusCache: FileStatusCache = NoopCache,
- userSpecifiedPartitionSpec: Option[PartitionSpec] = None,
- metadataOpsTimeNs: Option[Long] = None,
- override val partitionSchema: StructType)
+ fileStatusCache: FileStatusCache)
extends InMemoryFileIndex(
sparkSession,
- rootPathsSpecified,
+ rootPaths,
parameters,
userSpecifiedSchema,
- fileStatusCache,
- userSpecifiedPartitionSpec,
- metadataOpsTimeNs) {
+ fileStatusCache) {
+ def exposedLeafFiles: mutable.LinkedHashMap[Path,
_root_.org.apache.hadoop.fs.FileStatus] =
+ leafFiles
+ def exposedLeafDirToChildrenFiles: Map[Path,
Array[_root_.org.apache.hadoop.fs.FileStatus]] =
+ leafDirToChildrenFiles
+ }
+
+ override def refresh(): Unit = fileStatusCache.invalidateAll()
Review Comment:
This only invalidates the shared FileStatusCache. Once fullIndex has been
initialized, it still keeps its own cached leaf files, leaf-dir map, and
partition spec, while Spark's InMemoryFileIndex.refresh() also calls refresh0()
to rebuild those fields. I reproduced this by listing an index with pt=1,
creating pt=2, calling refresh(), and then listFiles(Nil, Nil) still returned
only pt=1. Please refresh/recreate fullIndex here so REFRESH TABLE and write
refresh paths do not leave unfiltered scans/allFiles/partitionSpec stale.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
