fix

Author: huan233usc

spark/src/main/scala/org/apache/spark/sql/delta/Snapshot.scala

@@ -30,7 +30,7 @@ import org.apache.spark.sql.delta.schema.SchemaUtils
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.stats.DataSkippingReader
 import org.apache.spark.sql.delta.stats.DataSkippingReaderConf
-import org.apache.spark.sql.delta.stats.DefaultDataSource
+import org.apache.spark.sql.delta.stats.DefaultStateProvider
 import org.apache.spark.sql.delta.stats.DeltaStatsColumnSpec
 import org.apache.spark.sql.delta.stats.StatisticsCollection
 import org.apache.spark.sql.delta.util.DeltaCommitFileProvider
@@ -228,15 +228,15 @@ class Snapshot(
   }
 
   /**
-   * The [[DefaultDataSource]] that owns the full state reconstruction pipeline:
+   * The [[DefaultStateProvider]] that owns the full state reconstruction pipeline:
    *   loadActions -> canonicalize -> repartition -> replay -> cache state
    *   -> extract add files -> parse stats -> cache stats
    *
    * V1's `stateDS`, `stateDF`, `allFiles`, and `withStats` all delegate here.
    * This replaces the previous `stateReconstruction` + `cachedState` chain.
    */
-  private[delta] lazy val dataSource: DefaultDataSource = {
-    new DefaultDataSource(
+  private[delta] lazy val stateProvider: DefaultStateProvider = {
+    new DefaultStateProvider(
       loadActions = () => loadActions,
       numPartitions = getNumPartitions,
       canonicalizeUdf = c => deltaLog.getCanonicalPathUdf()(c),
@@ -374,12 +374,12 @@ class Snapshot(
 
   /** The current set of actions in this [[Snapshot]] as plain Rows */
   def stateDF: DataFrame = recordFrameProfile("Delta", "stateDF") {
-    dataSource.stateDF
+    stateProvider.stateDF
   }
 
   /** The current set of actions in this [[Snapshot]] as a typed Dataset. */
   def stateDS: Dataset[SingleAction] = recordFrameProfile("Delta", "stateDS") {
-    dataSource.stateDS
+    stateProvider.stateDS
   }
 
   private[delta] def allFilesViaStateReconstruction: Dataset[AddFile] = {
@@ -478,8 +478,8 @@ class Snapshot(
       }
   }
 
-  // NOTE: stateReconstruction logic has been moved into DefaultDataSource.
-  // See [[dataSource]] which owns the full pipeline:
+  // NOTE: stateReconstruction logic has been moved into DefaultStateProvider.
+  // See [[stateProvider]] which owns the full pipeline:
   //   loadActions -> canonicalize -> repartition -> replay -> cache
 
   /**

spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala

@@ -264,11 +264,11 @@ trait DataSkippingReaderBase
 
   /**
    * Creates the Snapshot-level [[DefaultScanPlanner]] backed by
-   * [[Snapshot.dataSource]] which owns the full pipeline:
+   * [[Snapshot.stateProvider]] which owns the full pipeline:
    *   loadActions -> state reconstruction -> extract add -> parse stats -> cache
    */
   private[delta] def createPlanner(): DefaultScanPlanner = {
-    new DefaultScanPlanner(dataSource = snapshotToScan.dataSource)
+    new DefaultScanPlanner(stateProvider = snapshotToScan.stateProvider)
   }
 
   /** All files with the statistics column dropped completely. */

spark/src/main/scala/org/apache/spark/sql/delta/stats/ScanInterfaces.scala

@@ -27,17 +27,17 @@ import org.apache.spark.sql.delta.stats.DataFiltersBuilderUtils.ScanPipelineResu
 import org.apache.spark.sql.functions.{col, lit}
 import org.apache.spark.sql.types.StructType
 
-// ===================== Data Source Interface + Default =====================
+// ===================== State Provider Interface + Default =====================
 
 /**
  * Provides raw and parsed AddFile DataFrames for scan planning.
  *
  * Implementations own the full pipeline:
  *   raw source -> state reconstruction -> extract add files -> parse stats -> cache
  *
- * [[DefaultDataSource]] is the shared implementation used by both V1 and V2.
+ * [[DefaultStateProvider]] is the shared implementation used by both V1 and V2.
  */
-private[delta] trait ScanDataSource {
+private[delta] trait DeltaStateProvider {
 
   /** Flat AddFile rows with stats as a JSON string. */
   def allAddFiles(): DataFrame
@@ -50,7 +50,7 @@ private[delta] trait ScanDataSource {
 }
 
 /**
- * Shared [[ScanDataSource]] implementation used by both V1 and V2.
+ * Shared [[DeltaStateProvider]] implementation used by both V1 and V2.
  *
  * Owns the full pipeline from raw log actions to cached AddFile
  * DataFrames with parsed statistics:
@@ -68,7 +68,7 @@ private[delta] trait ScanDataSource {
  *  - V2: passes `DistributedLogReplayHelper.loadActions`,
  *    `callUDF("canonicalizePath", _)`, no retention, no caching.
  *
- * V1's `Snapshot` owns a `DefaultDataSource` instance and delegates
+ * V1's `Snapshot` owns a `DefaultStateProvider` instance and delegates
  * `stateDS`, `stateDF`, `allFiles` to it. This eliminates the
  * separate `stateReconstruction` / `cachedState` chain in `Snapshot`.
  *
@@ -96,7 +96,7 @@ private[delta] trait ScanDataSource {
  *                               `df => cacheDS(df, name).getDS`;
  *                               V2 passes None.
  */
-private[delta] class DefaultDataSource(
+private[delta] class DefaultStateProvider(
     loadActions: () => DataFrame,
     numPartitions: Int,
     canonicalizeUdf: Column => Column,
@@ -105,7 +105,7 @@ private[delta] class DefaultDataSource(
     minSetTransactionRetentionTimestamp: Option[Long] = None,
     stateCacheFactory: Option[Dataset[SingleAction] => Dataset[SingleAction]] = None,
     statsParseCacheFactory: Option[DataFrame => DataFrame] = None
-) extends ScanDataSource {
+) extends DeltaStateProvider {
 
   /**
    * Full state after log replay, optionally cached.
@@ -214,7 +214,7 @@ private[delta] trait DeltaScanPlanner {
 
   /**
    * Returns the AddFile DataFrame with parsed stats (and optionally cached).
-   * Delegates to the underlying [[ScanDataSource]].
+   * Delegates to the underlying [[DeltaStateProvider]].
    */
   def withParsedStats: DataFrame
 
@@ -379,20 +379,20 @@ private[delta] class DefaultScanPredicateBuilder(
 /**
  * Default shared implementation of [[DeltaScanPlanner]].
  *
- * Works for both V1 and V2 connectors. Delegates to [[ScanDataSource]]
+ * Works for both V1 and V2 connectors. Delegates to [[DeltaStateProvider]]
  * for the data pipeline (state reconstruction, stats parsing, caching).
  *
- * @param dataSource  Provides parsed + cached AddFile DataFrames
+ * @param stateProvider  Provides parsed + cached AddFile DataFrames
  */
 private[delta] class DefaultScanPlanner(
-    dataSource: ScanDataSource
+    stateProvider: DeltaStateProvider
 ) extends DeltaScanPlanner {
 
   /**
    * Parsed (and optionally cached) AddFile DataFrame.
-   * Delegates to [[ScanDataSource.withParsedStats]].
+   * Delegates to [[DeltaStateProvider.withParsedStats]].
    */
-  override def withParsedStats: DataFrame = dataSource.withParsedStats
+  override def withParsedStats: DataFrame = stateProvider.withParsedStats
 
   override def plan(
       filters: Seq[Expression],

spark/v2/src/main/java/io/delta/spark/internal/v2/read/DistributedLogReplayHelper.java

@@ -44,7 +44,7 @@
 public class DistributedLogReplayHelper {
 
   // Use the same column names as V1 (DeltaLogFileIndex.COMMIT_VERSION_COLUMN,
-  // Snapshot.ADD_STATS_TO_USE_COL_NAME) so DefaultDataSource works identically.
+  // Snapshot.ADD_STATS_TO_USE_COL_NAME) so DefaultStateProvider works identically.
   private static final String COMMIT_VERSION_COLUMN =
       org.apache.spark.sql.delta.DeltaLogFileIndex.COMMIT_VERSION_COLUMN();
   private static final String ADD_STATS_TO_USE_COL =
@@ -56,8 +56,8 @@ public class DistributedLogReplayHelper {
    * {@code add_stats_to_use} columns (same names as V1).
    *
    * <p>This is the V2 equivalent of V1's {@code Snapshot.loadActions}. It is consumed by {@link
-   * org.apache.spark.sql.delta.stats.DefaultDataSource} which handles the full state reconstruction
-   * pipeline.
+   * org.apache.spark.sql.delta.stats.DefaultStateProvider} which handles the full state
+   * reconstruction pipeline.
    *
    * @param spark SparkSession
    * @param snapshot Kernel snapshot (must be SnapshotImpl)
@@ -85,7 +85,7 @@ public static Dataset<Row> loadActionsV2(SparkSession spark, Snapshot snapshot)
    * Performs distributed log replay following V1's Snapshot.stateReconstruction algorithm.
    *
    * <p><b>Note:</b> For batch scans, prefer {@link #loadActionsV2} + {@link
-   * org.apache.spark.sql.delta.stats.DefaultDataSource} which shares the full pipeline with V1.
+   * org.apache.spark.sql.delta.stats.DefaultStateProvider} which shares the full pipeline with V1.
    * This method is retained for streaming use cases.
    *
    * @param spark SparkSession

spark/v2/src/main/java/io/delta/spark/internal/v2/read/DistributedScanBuilder.java

@@ -32,7 +32,7 @@
 import org.apache.spark.sql.delta.stats.DataFiltersBuilderV2;
 import org.apache.spark.sql.delta.stats.DefaultScanPlanner;
 import org.apache.spark.sql.delta.stats.DefaultScanPredicateBuilder;
-import org.apache.spark.sql.delta.stats.ScanDataSource;
+import org.apache.spark.sql.delta.stats.DeltaStateProvider;
 import org.apache.spark.sql.sources.Filter;
 
 /**
@@ -104,8 +104,8 @@ public ScanBuilder withReadSchema(StructType readSchema) {
    * Builds the scan using the shared {@link DefaultScanPlanner}:
    *
    * <ol>
-   *   <li>Create V2 {@link ScanDataSource}: loadActions -> state reconstruction -> extract add ->
-   *       parse stats (full pipeline, same class as V1)
+   *   <li>Create V2 {@link DeltaStateProvider}: loadActions -> state reconstruction -> extract add
+   *       -> parse stats (full pipeline, same class as V1)
    *   <li>Create shared {@link DefaultScanPlanner} backed by the data source
    *   <li>Create shared {@link DefaultScanPredicateBuilder} with V2-specific stat column paths
    *   <li>Convert Spark {@code Filter[]} to resolved Catalyst Expressions
@@ -116,14 +116,14 @@ public ScanBuilder withReadSchema(StructType readSchema) {
   @Override
   public Scan build() {
     // Step 1: V2 data source (full pipeline: loadActions -> reconstruct -> extract -> parse stats)
-    ScanDataSource v2DataSource =
+    DeltaStateProvider v2StateProvider =
         DataFiltersBuilderV2.createDataSource(
             () -> DistributedLogReplayHelper.loadActionsV2(spark, snapshot),
             numPartitions,
             snapshot);
 
     // Step 2: Shared planner backed by the data source
-    DefaultScanPlanner planner = DataFiltersBuilderV2.createPlanner(v2DataSource);
+    DefaultScanPlanner planner = DataFiltersBuilderV2.createPlanner(v2StateProvider);
 
     // Step 3: Shared predicate builder (V2-specific stat column paths injected)
     DefaultScanPredicateBuilder predicateBuilder =

spark/v2/src/main/scala/org/apache/spark/sql/delta/stats/DataFiltersBuilderV2.scala

@@ -134,7 +134,7 @@ object DataFiltersBuilderV2 {
   }
 
   /**
-   * Creates a [[DefaultDataSource]] for V2.
+   * Creates a [[DefaultStateProvider]] for V2.
    *
    * The data source owns the full pipeline:
    *   loadActions -> state reconstruction -> extract add -> parse stats
@@ -143,15 +143,15 @@ object DataFiltersBuilderV2 {
    * @param loadActions    Supplier for the union of checkpoint + delta files
    * @param numPartitions  Number of partitions for state reconstruction
    * @param snapshot       Kernel snapshot (provides table schema for stats)
-   * @return A [[DefaultDataSource]] ready for V2 use
+   * @return A [[DefaultStateProvider]] ready for V2 use
    */
   def createDataSource(
       loadActions: () => DataFrame,
       numPartitions: Int,
-      snapshot: Snapshot): DefaultDataSource = {
+      snapshot: Snapshot): DefaultStateProvider = {
     val tableSchema = getSparkTableSchema(snapshot)
     val statsSchema = DataFiltersBuilderUtils.buildStatsSchema(tableSchema)
-    new DefaultDataSource(
+    new DefaultStateProvider(
       loadActions = loadActions,
       numPartitions = numPartitions,
       canonicalizeUdf = c => callUDF("canonicalizePath", c),
@@ -163,11 +163,11 @@ object DataFiltersBuilderV2 {
   /**
    * Creates a [[DefaultScanPlanner]] for V2.
    *
-   * @param dataSource  V2 [[ScanDataSource]] (from [[createDataSource]])
+   * @param stateProvider  V2 [[DeltaStateProvider]] (from [[createDataSource]])
    * @return A [[DefaultScanPlanner]] ready for V2 use
    */
-  def createPlanner(dataSource: ScanDataSource): DefaultScanPlanner = {
-    new DefaultScanPlanner(dataSource = dataSource)
+  def createPlanner(stateProvider: DeltaStateProvider): DefaultScanPlanner = {
+    new DefaultScanPlanner(stateProvider = stateProvider)
   }
 
   /**