[SPARK-59149][SQL] Use isEmpty/nonEmpty instead of size comparisons in the SQL modules

[uros-b]

What changes were proposed in this pull request?

This replaces size-vs-zero comparisons with the direct emptiness / non-emptiness predicates across the SQL modules (catalyst, core, hive, connect), in both main and test sources:

• Emptiness: x.size == 0 / x.size() == 0 -> x.isEmpty.
• Non-emptiness: x.size > 0 / x.size >= 1 -> x.nonEmpty for Scala collections, and !x.isEmpty for Java collections (java.util.*, protobuf lists, ByteString, etc.), which have no nonEmpty.

Touched files span Analyzer, predicates, AstBuilder (catalyst); DataSource, resources, InferVariantShreddingSchema, ParquetPartitionReaderFactory, AsyncLogPurge, AsyncProgressTrackingMicroBatchExecution, HDFSBackedStateStoreProvider (core main); OrcUtils, TableReader, OrcFileOperator (hive main); MLCache, SparkConnectPlanner (connect main); plus the corresponding test suites.

Deliberately left unchanged, because they are not collection-emptiness checks:

• CostBasedJoinReorder -- planCost.size is a numeric BigInt field on case class Cost(card, size).
• SubExprEvaluationRuntimeSuite, FetchErrorDetailsHandlerSuite -- the receivers are Guava caches (LoadingCache / CacheBuilder), which expose size() but not isEmpty().
• On the AsyncProgressTrackingMicroBatchExecution line, getActiveCount > 0 (a thread-pool active-thread count) is left as-is; only the adjacent getQueue.size() check is converted.

Why are the changes needed?

isEmpty / nonEmpty state the intent directly and are the idioms used elsewhere in these files. They also avoid computing a full size only to compare it against zero (for some collections size is O(n) while isEmpty is O(1)). Behavior is unchanged: for every converted receiver -- Scala collections, java.util collections, StructType, and protobuf/ByteString types -- the predicate is exactly equivalent to the original comparison.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests. This is a behavior-preserving refactor; the catalyst, sql (core, main + test), hive, and connect modules compile cleanly.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8)

[uros-b]

Waiting for CI

[dongjoon-hyun]

Could you try to find more instances, @uros-b ?

$ git grep '\.size == 0' | grep scala
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetRangeCalculator.scala:   * Empty (`KafkaOffsetRange.size == 0`) or invalid (`KafkaOffsetRange.size < 0`) ranges  will be
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:          val in = if (buf.size == 0) {
core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala:      assert(tracker.shuffleStatuses(0).mapIdToMapIndex.filter(_._2 == 0).size == 0)
core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala:      assert(tracker.shuffleStatuses(0).mapIdToMapIndex.filter(_._2 == 0).size == 0)
core/src/test/scala/org/apache/spark/SparkTestSuite.scala:    if (loggers.size == 0) {
core/src/test/scala/org/apache/spark/deploy/StandaloneDynamicAllocationSuite.scala:      assert(beforeList.intersect(afterList).size == 0)
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:    assert(sched.speculativeTasks.size == 0)
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:      assert(sched.speculativeTasks.size == 0)
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:    assert(sched.speculativeTasks.size == 0)
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:    assert(sched.speculativeTasks.size == 0)
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:    assert(master.getLocations("a1").size == 0, "a1 was not removed from master")
core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala:    assert(master.getLocations("a1").size == 0, "a1 was not removed from master")
core/src/test/scala/org/apache/spark/util/NextIteratorSuite.scala:      if (ints.size == 0) {
mllib/src/main/scala/org/apache/spark/ml/tree/treeModels.scala:      assert(totalImportances.size == 0, s"Unknown error in computing feature" +
mllib/src/main/scala/org/apache/spark/mllib/stat/test/ChiSqTest.scala:    val expArr = if (expected.size == 0) Array.tabulate(size)(_ => 1.0 / size) else expected.toArray
mllib/src/main/scala/org/apache/spark/mllib/stat/test/ChiSqTest.scala:    val expSum = if (expected.size == 0.0) 1.0 else expArr.sum
mllib/src/test/scala/org/apache/spark/ml/recommendation/ALSSuite.scala:    assert(shuffledUserFactors.size == 0)
mllib/src/test/scala/org/apache/spark/ml/recommendation/ALSSuite.scala:    assert(shuffledItemFactors.size == 0)
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala:      if (other.planCost.card == 0 || other.planCost.size == 0) {
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionSetSuite.scala:    assert((initialSet - (aUpper + 1)).size == 0)
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ExpressionSetSuite.scala:    assert((initialSet - (aLower + 1)).size == 0)
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/SubexpressionEliminationSuite.scala:    assert(equivalence.getAllExprStates(1).size == 0)
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala:      schema.size == 0 || schema.exists {
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala:      if (schema.getFieldNames.size == 0) {
sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala:    assert(plan1.collect { case p: BroadcastHashJoinExec => p }.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/OptimizeMetadataOnlyQuerySuite.scala:    assert(localRelations.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala:      assert(exchanges.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala:      assert(exchanges.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala:            assert(outputOrdering.head.sameOrderExpressions.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala:            assert(outputOrdering.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala:    assert(outputOrdering.head.sameOrderExpressions.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala:    assert(outputOrdering3.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala:    assert(outputOrdering.head.sameOrderExpressions.size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:      assert(findTopLevelUnion(adaptivePlan).size == 0)
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala:            assert(reusedSubqueryIds.size == 0,
sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala:    assert ((modifiedParameters -- parameters.keys).size == 0)
sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala:    if (hivePartitionRDDs.size == 0) {

[uros-b]

It's a trade-off I guess... too many occurrences across different modules vs. too narrow change which results in very small PR. Came across this one intially, but will definitely look up some more instances and expand accordingly, thank you @dongjoon-hyun! let's not merge this yet

[HyukjinKwon]

LGTM but I believe there are a lot of instances like this. Let's probably try to avoid fixing this alone in a PR.

[uros-b]

Okay, here's the plan - I extended the current PR to cover a wider range of occurrences under the SQL component.

Of course please note the caution here, as some .size == 0 are numeric field comparisons, not collection emptiness.
For example, Dongjoon's list above includes CostBasedJoinReorder's planCost.size == 0, where size is a cost field.
Also, please note that Java collections have no .nonEmpty, so I needed to use !.isEmpty instead in a few cases.

In any case, for other components, let's visit those in separate PRs, such as:

• CORE: [SPARK-59172][CORE] Use isEmpty/nonEmpty instead of size comparisons in core #58470
• MLLIB: [SPARK-59173][ML] Use isEmpty/nonEmpty instead of size comparisons in MLlib #58471