JingsongLi commented on code in PR #9298:
URL: https://github.com/apache/paimon/pull/9298#discussion_r3828126439
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java:
##########
@@ -81,52 +107,92 @@ public List<PartitionStatistics> collect(List<Map<String,
String>> partitions) {
if (partitions.isEmpty()) {
return Collections.emptyList();
}
- int threads = Math.min(parallelism, partitions.size());
+ SimpleStatsExtractor rowCounter = withRecordCount ? rowCounter() :
null;
+ if (withRecordCount && rowCounter == null) {
+ LOG.info(
+ "No row counter could be built for format {} of table {},
so the row counts of "
+ + "the measured partitions stay unknown.",
+ table.format(),
+ table.fullName());
+ }
+ // A listing is one request per partition, a footer read one per file,
so counting rows
+ // leaves work to spread even when a single partition was asked for.
+ int threads = rowCounter == null ? Math.min(parallelism,
partitions.size()) : parallelism;
if (threads == 1) {
List<PartitionStatistics> statistics = new
ArrayList<>(partitions.size());
for (Map<String, String> partition : partitions) {
- statistics.add(measure(partition));
+ List<FileStatus> files = listDataFiles(partition);
+ List<Long> rowCounts = new ArrayList<>(files.size());
+ for (FileStatus file : files) {
+ if (rowCounter != null) {
+ rowCounts.add(rowCount(rowCounter, file));
+ }
+ }
+ statistics.add(statistics(partition, files, sum(rowCounts,
rowCounter != null)));
}
return statistics;
}
ExecutorService executor =
ThreadPoolUtils.createCachedThreadPool(threads,
"FORMAT-TABLE-STATS-THREAD-POOL");
try {
- List<Future<PartitionStatistics>> futures = new
ArrayList<>(partitions.size());
+ List<Future<List<FileStatus>>> listings = new
ArrayList<>(partitions.size());
for (Map<String, String> partition : partitions) {
- futures.add(executor.submit(() -> measure(partition)));
+ listings.add(executor.submit(() -> listDataFiles(partition)));
+ }
+ List<List<FileStatus>> files = new ArrayList<>(partitions.size());
+ for (Future<List<FileStatus>> listing : listings) {
+ files.add(await(listing));
+ }
+ // Every file of every partition goes to the same pool, so one
partition holding many
+ // files is counted with all of it rather than with one thread of
it.
+ List<List<Future<Long>>> rowCounts = new
ArrayList<>(partitions.size());
+ for (List<FileStatus> partitionFiles : files) {
+ List<Future<Long>> counts = new
ArrayList<>(partitionFiles.size());
+ for (FileStatus file : partitionFiles) {
+ if (rowCounter != null) {
+ counts.add(executor.submit(() -> rowCount(rowCounter,
file)));
Review Comment:
[P2] Please run full-ANALYZE footer reads on Spark executors. This
parallelizes files, but every footer open still runs in a driver-local pool,
and the driver eagerly retains one `FileStatus` plus one `Future` for every
data file before aggregation. A format table with hundreds of thousands or
millions of files can therefore make ANALYZE driver-bound or exhaust the driver
heap while executor capacity is idle. The Spark layer could batch file
descriptors and process them with RDD `mapPartitions`, creating `FileIO` and
the stats extractor once per task and reducing partial results per catalog
partition; `format-table.statistics.parallelism` can bound the RDD
partitions/storage concurrency. The engine-neutral footer parsing and merge
logic can remain in `paimon-core`, and NOSCAN can keep the local path.
--
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]
