[SPARK-55978][SQL] Address review comments for TABLESAMPLE SYSTEM

Summary:
CONTEXT: Review feedback on PR apache#54972 for TABLESAMPLE SYSTEM block sampling.
WHAT:
- Add DSv2 SampleMethod Java enum to replace boolean isSystemSampling parameter
- Update SupportsPushDownTableSample Javadoc to clarify BERNOULLI vs SYSTEM
- Replace _LEGACY_ERROR_TEMP_0035 with UNSUPPORTED_FEATURE.TABLESAMPLE_SYSTEM
- Restrict SYSTEM sample pushdown to direct table scans only (no subqueries/views)
- Remove findScanBuilderHolder that incorrectly traversed through Filter/Project

Test Plan:
Existing tests. Error class added to error-conditions.json.

Author: Stanley Yao

common/utils/src/main/resources/error/error-conditions.json

@@ -8091,6 +8091,11 @@
         "message" : [
           "The target table is <table>."
         ]
+      },
+      "TABLESAMPLE_SYSTEM" : {
+        "message" : [
+          "TABLESAMPLE SYSTEM (block sampling) requires a DSv2 data source that supports sample pushdown (SupportsPushDownTableSample)."
+        ]
       }
     },
     "sqlState" : "42902"

sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SampleMethod.java

@@ -0,0 +1,33 @@
+/*
+ * 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.connector.read;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * The sampling method for TABLESAMPLE.
+ *
+ * @since 4.1.0
+ */
+@Evolving
+public enum SampleMethod {
+  /** Row-level sampling (BERNOULLI). Each row is independently selected. */
+  BERNOULLI,
+  /** Block-level sampling (SYSTEM). Entire partitions/splits are included or skipped. */
+  SYSTEM
+}

sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownTableSample.java

@@ -29,7 +29,7 @@
 public interface SupportsPushDownTableSample extends ScanBuilder {
 
   /**
-   * Pushes down SAMPLE to the data source.
+   * Pushes down BERNOULLI (row-level) SAMPLE to the data source.
    */
   boolean pushTableSample(
       double lowerBound,
@@ -38,18 +38,15 @@ boolean pushTableSample(
       long seed);
 
   /**
-   * Pushes down SAMPLE to the data source with sample method awareness.
-   * Data sources can override this to distinguish SYSTEM (block) from BERNOULLI (row) sampling.
-   * By default, rejects SYSTEM sampling for backward compatibility and delegates BERNOULLI to
-   *   the 4-parameter version.
+   * Pushes down SAMPLE to the data source with the specified sampling method.
    */
   default boolean pushTableSample(
       double lowerBound,
       double upperBound,
       boolean withReplacement,
       long seed,
-      boolean isSystemSampling) {
-    if (isSystemSampling) {
+      SampleMethod sampleMethod) {
+    if (sampleMethod == SampleMethod.SYSTEM) {
       // If the data source hasn't overridden this method, it must have not added support
       // for SYSTEM sampling. Don't apply sample pushdown.
       return false;

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

@@ -1046,10 +1046,8 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
           // not able to support block/split sampling. Now it fell through to the row-based
           // sampling. Therefore error out.
           throw new AnalysisException(
-            errorClass = "_LEGACY_ERROR_TEMP_0035",
-            messageParameters = Map("message" ->
-              ("TABLESAMPLE SYSTEM (block sampling) must be pushed down to a DSv2 data source. " +
-              "It cannot be executed as row-level sampling.")))
+            errorClass = "UNSUPPORTED_FEATURE.TABLESAMPLE_SYSTEM",
+            messageParameters = Map.empty)
         }
         execution.SampleExec(lb, ub, withReplacement, seed, planLater(child)) :: Nil
       case logical.LocalRelation(output, data, _, stream) =>

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala

@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.connector.expressions.{IdentityTransform, SortOrder}
 import org.apache.spark.sql.connector.expressions.filter.Predicate
-import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownLimit, SupportsPushDownOffset, SupportsPushDownRequiredColumns, SupportsPushDownTableSample, SupportsPushDownTopN, SupportsPushDownV2Filters}
+import org.apache.spark.sql.connector.read.{SampleMethod => SampleMethodV2, Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownLimit, SupportsPushDownOffset, SupportsPushDownRequiredColumns, SupportsPushDownTableSample, SupportsPushDownTopN, SupportsPushDownV2Filters}
 import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, DataSourceUtils}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.connector.{PartitionPredicateField, PartitionPredicateImpl, SupportsPushDownCatalystFilters}
@@ -298,7 +298,10 @@ object PushDownUtils extends Logging {
       case s: SupportsPushDownTableSample =>
         s.pushTableSample(
           sample.lowerBound, sample.upperBound, sample.withReplacement, sample.seed,
-          sample.sampleMethod == SampleMethod.System)
+          sample.sampleMethod match {
+            case SampleMethod.Bernoulli => SampleMethodV2.BERNOULLI
+            case SampleMethod.System => SampleMethodV2.SYSTEM
+          })
       case _ => false
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala

@@ -812,42 +812,30 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
       }
   }
 
-  private def findScanBuilderHolder(plan: LogicalPlan): Option[ScanBuilderHolder] = {
-    plan match {
-      case s: ScanBuilderHolder => Some(s)
-      case s: SubqueryAlias => findScanBuilderHolder(s.child)
-      case p: Project => findScanBuilderHolder(p.child)
-      case f: Filter => findScanBuilderHolder(f.child)
-      case _ => None
-    }
-  }
-
   def pushDownSample(plan: LogicalPlan): LogicalPlan = plan.transform {
     case sample: Sample if sample.sampleMethod == SampleMethod.System =>
-      findScanBuilderHolder(sample.child) match {
-        case Some(sHolder) =>
-          val tableSample = TableSampleInfo(
-            sample.lowerBound,
-            sample.upperBound,
-            sample.withReplacement,
-            sample.seed,
-            sampleMethod = sample.sampleMethod)
-          val pushed = PushDownUtils.pushTableSample(sHolder.builder, tableSample)
-          if (pushed) {
-            sHolder.pushedSample = Some(tableSample)
-            sample.child
-          } else {
-            throw new AnalysisException(
-              errorClass = "_LEGACY_ERROR_TEMP_0035",
-              messageParameters = Map("message" ->
-                ("TABLESAMPLE SYSTEM requires a data source that supports " +
-                "table sample pushdown (SupportsPushDownTableSample).")))
-          }
-        case None =>
+      val sHolder = sample.child match {
+        case s: ScanBuilderHolder => s
+        case SubqueryAlias(_, s: ScanBuilderHolder) => s
+        case _ =>
           throw new AnalysisException(
-            errorClass = "_LEGACY_ERROR_TEMP_0035",
-            messageParameters = Map("message" ->
-              "TABLESAMPLE SYSTEM is only supported for DSv2 data source scan relations."))
+            errorClass = "UNSUPPORTED_FEATURE.TABLESAMPLE_SYSTEM",
+            messageParameters = Map.empty)
+      }
+      val tableSample = TableSampleInfo(
+        sample.lowerBound,
+        sample.upperBound,
+        sample.withReplacement,
+        sample.seed,
+        sampleMethod = sample.sampleMethod)
+      val pushed = PushDownUtils.pushTableSample(sHolder.builder, tableSample)
+      if (pushed) {
+        sHolder.pushedSample = Some(tableSample)
+        sample.child
+      } else {
+        throw new AnalysisException(
+          errorClass = "UNSUPPORTED_FEATURE.TABLESAMPLE_SYSTEM",
+          messageParameters = Map.empty)
       }
 
     case sample: Sample => sample.child match {