[Bug] The Spark query on Paimon consumes a relatively large amount of memory, and there are also errors in field type conversion.

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.

### Paimon version

paimon：1.1.1

### Compute Engine

flink:1.18.1
spark 3.5.1

### Minimal reproduce step

Flink writes data to table T_1 in real time. The 'bucket' of table T_1 is '80', and 'snapshot. time-retained' is '12 h'. Each partition has approximately 6 billion datas per day, and the distribution is relatively even. When I use the following SparkSQL to count the number of data entries in each partition：

```sql
create table data_detail(
.....
warning string
brand string,
dt INT
)PARTITIONED BY (brand,dt)
with(
'bucket' = '80',
'bucket-key' = 'vin',
'write-buffer-spillable' = 'true',
'write-buffer-for-append' = 'true',
'sink.parallelism' = '80',
'snapshot.time-retained' = '12 h',
'file.format' = 'parquet',
'write-buffer-size' = '1024 MB',
'write-manifest-cache' = '256 MB',
'target-file-size' = '512 MB',
'scan.plan-sort-partition' = 'true',
'partition.expiration-time' = '735 d',
'partition.expiration-check-interval' = '1 d',
'partition.timestamp-formatter' = 'yyyyMMdd'
);
```

```
select dt,count(1) from paimon.test.data_detail group by dt,brand
```

1. If spark.driver.memory is set to 6G, the following error will occur:
Why does counting the number of records based on the date partition field require so much memory resources?

```
java.lang.RuntimeException: java.util.concurrent.ExecutionException: java.lang.OutOfMemoryError: Java heap space
at org.apache.paimon.utils.ThreadPoolUtils$2.next(ThreadPoolUtils.java:176)
at org.apache.paimon.utils.ThreadPoolUtils$2.next(ThreadPoolUtils.java:162)
at org.apache.paimon.shade.guava30.com.google.common.collect.Iterators$ConcatenatedIterator.hasNext(Iterators.java:1333)
at org.apache.paimon.operation.AbstractFileStoreScan.plan(AbstractFileStoreScan.java:263)
at org.apache.paimon.table.source.snapshot.SnapshotReaderImpl.read(SnapshotReaderImpl.java:318)
at org.apache.paimon.table.source.snapshot.ReadPlanStartingScanner.scan(ReadPlanStartingScanner.java:45)
at org.apache.paimon.table.source.DataTableBatchScan.plan(DataTableBatchScan.java:77)
at org.apache.paimon.spark.PaimonScanBuilder.pushAggregation(PaimonScanBuilder.scala:121)
at org.apache.paimon.spark.PaimonScanBuilder.supportCompletePushDown(PaimonScanBuilder.scala:95)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$.org$apache$spark$sql$execution$datasources$v2$V2ScanRelationPushDown$$rewriteAggregate(V2ScanRelationPushDown.scala:122)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$$anonfun$pushDownAggregates$1.applyOrElse(V2ScanRelationPushDown.scala:97)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$$anonfun$pushDownAggregates$1.applyOrElse(V2ScanRelationPushDown.scala:95)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:461)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(origin.scala:76)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:461)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:32)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:32)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:32)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:437)
at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:405)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$.pushDownAggregates(V2ScanRelationPushDown.scala:95)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$.$anonfun$apply$4(V2ScanRelationPushDown.scala:46)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$.$anonfun$apply$8(V2ScanRelationPushDown.scala:52)
at scala.collection.LinearSeqOptimized.foldLeft(LinearSeqOptimized.scala:126)
at scala.collection.LinearSeqOptimized.foldLeft$(LinearSeqOptimized.scala:122)
at scala.collection.immutable.List.foldLeft(List.scala:91)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$.apply(V2ScanRelationPushDown.scala:51)
at org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown$.apply(V2ScanRelationPushDown.scala:38)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:222)
at scala.collection.LinearSeqOptimized.foldLeft(LinearSeqOptimized.scala:126)
at scala.collection.LinearSeqOptimized.foldLeft$(LinearSeqOptimized.scala:122)
at scala.collection.immutable.List.foldLeft(List.scala:91)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:219)
at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:211)
at scala.collection.immutable.List.foreach(List.scala:431)
```

2. If I increase the Spark driver memory, the query becomes extremely slow. After over ten minutes or even half an hour, the following error occurs:

```
Caused by: java.lang.ClassCastException: org.apache.paimon.data.BinaryString cannot be cast to java.lang.Integer
at org.apache.paimon.data.GenericRow.getInt(GenericRow.java:152)
at org.apache.paimon.spark.AbstractSparkInternalRow.getInt(AbstractSparkInternalRow.java:120)
at org.apache.spark.sql.catalyst.expressions.JoinedRow.getInt(JoinedRow.scala:92)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
at org.apache.spark.sql.execution.LocalTableScanExec.$anonfun$unsafeRows$1(LocalTableScanExec.scala:44)
at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
at scala.collection.TraversableLike.map(TraversableLike.scala:286)
at scala.collection.TraversableLike.map$(TraversableLike.scala:279)
at scala.collection.AbstractTraversable.map(Traversable.scala:108)
at org.apache.spark.sql.execution.LocalTableScanExec.unsafeRows$lzycompute(LocalTableScanExec.scala:44)
at org.apache.spark.sql.execution.LocalTableScanExec.unsafeRows(LocalTableScanExec.scala:39)
at org.apache.spark.sql.execution.LocalTableScanExec.rdd$lzycompute(LocalTableScanExec.scala:53)
at org.apache.spark.sql.execution.LocalTableScanExec.rdd(LocalTableScanExec.scala:48)
at org.apache.spark.sql.execution.LocalTableScanExec.inputRDD(LocalTableScanExec.scala:97)
at org.apache.spark.sql.execution.InputRDDCodegen.inputRDDs(WholeStageCodegenExec.scala:455)
at org.apache.spark.sql.execution.InputRDDCodegen.inputRDDs$(WholeStageCodegenExec.scala:454)
at org.apache.spark.sql.execution.LocalTableScanExec.inputRDDs(LocalTableScanExec.scala:32)
at org.apache.spark.sql.execution.ProjectExec.inputRDDs(basicPhysicalOperators.scala:51)
at org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:751)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:195)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:246)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:243)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:191)
at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:364)
at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:445)
at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:4332)
at org.apache.spark.sql.Dataset.$anonfun$collect$1(Dataset.scala:3573)
at org.apache.spark.sql.Dataset.$anonfun$withAction$2(Dataset.scala:4322)
at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:546)
at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:4320)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:125)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:201)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:108)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:66)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:4320)
at org.apache.spark.sql.Dataset.collect(Dataset.scala:3573)
at org.apache.kyuubi.engine.spark.operation.ExecuteStatement.fullCollectResult(ExecuteStatement.scala:85)
at org.apache.kyuubi.engine.spark.operation.ExecuteStatement.collectAsIterator(ExecuteStatement.scala:203)
at org.apache.kyuubi.engine.spark.operation.ExecuteStatement.$anonfun$executeStatement$1(ExecuteStatement.scala:100)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.kyuubi.engine.spark.operation.SparkOperation.$anonfun$withLocalProperties$1(SparkOperation.scala:166)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:201)
at org.apache.kyuubi.engine.spark.operation.SparkOperation.withLocalProperties(SparkOperation.scala:150)
at org.apache.kyuubi.engine.spark.operation.ExecuteStatement.executeStatement(ExecuteStatement.scala:94)
... 6 more
```

3. When the order of the fields in the GROUP BY clause is changed, the execution can be successful.

```
select brand,dt,count(1) cnt from data_detail group by brand,dt;
```

### What doesn't meet your expectations?

1. Why would just counting the data volume in partitions require so much memory resources?

2. Why would a type conversion error occur?

### Anything else?

_No response_

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Reproduce the grouped query with the listed Paimon, Flink, and Spark versions, then start at PaimonScanBuilder.scala and follow the stack traces into AbstractFileStoreScan.java, GenericRow.java, and AbstractSparkInternalRow.java. Compare both GROUP BY field orders and the low-memory and higher-memory runs; done means the query avoids the reported driver OOM, completes without the BinaryString-to-Integer error, and behaves consistently for either order.

Written by the indexing model from the issue text.