nsivabalan commented on code in PR #20152: URL: https://github.com/apache/hudi/pull/20152#discussion_r4284110833
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableLayoutAnalyzer.java: ########## @@ -0,0 +1,1825 @@ +/* + * 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.hudi.utilities; + +import org.apache.hudi.avro.model.HoodieMetadataColumnStats; +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.hadoop.fs.HadoopFSUtils; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.storage.StorageConfiguration; +import org.apache.hudi.storage.StoragePath; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.UniformReservoir; +import lombok.extern.slf4j.Slf4j; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.spark.api.java.JavaSparkContext; + +import javax.annotation.Nullable; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Serializable; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.time.LocalDate; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.format.DateTimeParseException; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Analyzes the physical layout of a Hudi table: how bytes and base files are spread across + * partitions, how skewed that spread is, and whether the layout shows the classic symptoms + * that make queries and table services slow. + * + * <p>The tool is strictly read-only. It reads the timeline and the file listing (through the + * metadata table when the table has one) and prints a report; it never writes to the table. + * + * <h3>What it reports</h3> + * <ul> + * <li>Table-level base-file size distribution (count, total, min / max / mean / median, + * p50 / p90 / p95 / p99) and the per-partition file-count distribution.</li> + * <li>Partition-size skew: coefficient of variation, Gini coefficient, largest / top-10 + * partition share, and partitions above mean + 2 sigma.</li> + * <li>With {@code --enable-partition-stats}, one row per partition sorted by total bytes + * (capped with {@code --top-n}).</li> + * <li>With {@code --include-row-counts}, per-partition record counts, taken from the metadata + * table's column stats when present and from Parquet footers otherwise.</li> + * <li>With {@code --analyze-table-characteristics}, three detectors, each with a verdict, + * the thresholds it used, and actionable findings: micro-partitioning, small-file + * pile-up (CLEAN / MODERATE / SEVERE / SKIPPED), and hot partitions by recent write + * volume.</li> + * </ul> + * + * <p>Date filtering ({@code --num-days}, {@code --start-date}, {@code --end-date}) narrows the + * partitions considered and only works on tables whose partition names are of the form + * {@code [column=]yyyy-M-d} or {@code [column=]yyyy/M/d}; the half-open interval + * {@code [start, end)} is used. {@code --start-date} takes precedence over {@code --num-days}. + * + * <h3>Example</h3> + * <pre> + * spark-submit --master "local[2]" \ + * --class org.apache.hudi.utilities.HoodieTableLayoutAnalyzer \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-1.3.0-SNAPSHOT.jar \ + * --base-path s3://bucket/path/to/table \ + * --enable-partition-stats --analyze-table-characteristics --output JSON + * </pre> + */ +@Slf4j +public class HoodieTableLayoutAnalyzer implements Serializable { + + private static final long serialVersionUID = 1L; + + // Date formatter for parsing partition dates (example: 2023/5/5/ or 2023-5-5). + private static final DateTimeFormatter DATE_FORMATTER = + (new DateTimeFormatterBuilder()).appendOptional(DateTimeFormatter.ofPattern("yyyy/M/d")).appendOptional(DateTimeFormatter.ofPattern("yyyy-M-d")).toFormatter(); + + // File size stats will be displayed in the units specified below. + private static final String[] FILE_SIZE_UNITS = {"B", "KB", "MB", "GB", "TB"}; + + // The record key column is required and populated exactly once per row, so its value count + // in the metadata table's column stats equals the row count of the file. + private static final String ROW_COUNT_COLUMN = "_hoodie_record_key"; + + static final String MICRO_PARTITION_DETECTOR = "micro-partition"; + static final String SMALL_FILES_DETECTOR = "small-files"; + static final String HOT_PARTITIONS_DETECTOR = "hot-partitions"; + + // Spark context + private transient JavaSparkContext jsc; + // config + private final Config cfg; + // Properties with source, hoodie client, key generator etc. + private final TypedProperties props; + + public HoodieTableLayoutAnalyzer(JavaSparkContext jsc, Config cfg) { + this.jsc = jsc; + this.cfg = cfg; + + this.props = cfg.propsFilePath == null + ? UtilHelpers.buildProperties(cfg.configs) + : readConfigFromFileSystem(jsc, cfg); + } + + /** + * Reads config from the file system. + * + * @param jsc {@link JavaSparkContext} instance. + * @param cfg {@link Config} instance. + * @return the {@link TypedProperties} instance. + */ + private TypedProperties readConfigFromFileSystem(JavaSparkContext jsc, Config cfg) { + return UtilHelpers.readConfig(jsc.hadoopConfiguration(), new Path(cfg.propsFilePath), cfg.configs) + .getProps(true); + } + + public static class Config implements Serializable { + + private static final long serialVersionUID = 1L; + + @Parameter(names = {"--base-path", "-bp"}, description = "Base path for the table", required = false) + public String basePath = null; + + @Parameter(names = {"--num-days", "-nd"}, description = "Consider files modified within this many days.", required = false) + public long numDays = 0; + + @Parameter(names = {"--start-date", "-sd"}, description = "Consider files modified on or after this date.", required = false) + public String startDate = null; + + @Parameter(names = {"--end-date", "-ed"}, description = "Consider files modified before this date.", required = false) + public String endDate = null; + + @Parameter(names = {"--enable-table-stats", "-fs"}, description = "Show file-level stats.", required = false) + public boolean tableStats = false; + + @Parameter(names = {"--enable-partition-stats", "-ps"}, description = "Show partition-level stats.", required = false) + public boolean partitionStats = false; + + @Parameter(names = {"--output", "-o"}, description = "Output format: TABLE (default) or JSON.", required = false) + public String output = "TABLE"; + + @Parameter(names = {"--top-n", "-tn"}, + description = "When --enable-partition-stats is set, only show the top N largest partitions (sorted by totalBytes desc). Default 0 = show all.", + required = false) + public int topN = 0; + + @Parameter(names = {"--include-row-counts", "-irc"}, + description = "Opt-in: also collect per-partition row counts. Uses metadata table column stats when present, " + + "Parquet footer reads otherwise. Adds numRecords + avgRowSize columns.", + required = false) + public boolean includeRowCounts = false; + + @Parameter(names = {"--analyze-table-characteristics", "-atc"}, + description = "Opt-in: run micro-partition / small-file / hot-partition detectors and emit a verdict section.", + required = false) + public boolean analyzeTableCharacteristics = false; + + @Parameter(names = {"--print-top-k"}, + description = "When >0 and --analyze-table-characteristics is set, additionally print the top K " + + "partitions with the smallest avg file size under the micro and small-file detectors. Default 0 = off.", + required = false) + public int printTopK = 0; + + @Parameter(names = {"--micro-partition-count-threshold"}, + description = "Table is flagged micro-partitioned when numPartitions exceeds this. Default 10000.", + required = false) + public int microPartitionCountThreshold = 10000; + + @Parameter(names = {"--micro-partition-min-files"}, + description = "Per-partition file count gate for the size-based micro-partition rule. Default 25.", + required = false) + public int microPartitionMinFiles = 25; + + @Parameter(names = {"--micro-partition-max-avg-bytes"}, + description = "Per-partition avg-file-size threshold (bytes) for the size-based rule. Default 52428800 (50 MB).", + required = false) + public long microPartitionMaxAvgBytes = 50L * 1024 * 1024; + + @Parameter(names = {"--micro-partition-min-age-days"}, + description = "Minimum table age before applying the size-based micro-partition rule. Default 30.", + required = false) + public int microPartitionMinAgeDays = 30; + + @Parameter(names = {"--small-files-min-files-per-partition"}, + description = "Per-partition file count gate for small-files detection; partitions with fewer files are excluded from the prevalence ratio. Default 5.", + required = false) + public int smallFilesMinFilesPerPartition = 5; + + @Parameter(names = {"--small-files-threshold-bytes"}, + description = "Avg-file-size threshold (bytes) for flagging a qualifying partition. Default 52428800 (50 MB).", + required = false) + public long smallFilesThresholdBytes = 50L * 1024 * 1024; + + @Parameter(names = {"--small-files-moderate-pct"}, + description = "Fraction of qualifying partitions flagged to trigger MODERATE verdict. Default 0.10.", + required = false) + public double smallFilesModeratePct = 0.10; + + @Parameter(names = {"--small-files-severe-pct"}, + description = "Fraction of qualifying partitions flagged to trigger SEVERE verdict. Default 0.30.", + required = false) + public double smallFilesSeverePct = 0.30; + + @Parameter(names = {"--small-files-min-table-commits"}, + description = "Minimum total ingest commits (active+archived) before emitting the small-file verdict. Default 10.", + required = false) + public int smallFilesMinTableCommits = 10; + + @Parameter(names = {"--hot-window-commits"}, + description = "Number of recent ingest commits to scan for hot-partition detection. Default 50.", + required = false) + public int hotWindowCommits = 50; + + @Parameter(names = {"--hot-partition-commit-share"}, + description = "Minimum share of the hot window a partition must appear in to be flagged hot. Default 0.5.", + required = false) + public double hotPartitionCommitShare = 0.5; + + @Parameter(names = {"--props-path", "-pp"}, description = "Properties file containing base paths one per line", required = false) + public String propsFilePath = null; + + @Parameter(names = {"--parallelism", "-pl"}, description = "Parallelism for valuation", required = false) + public int parallelism = 200; + + @Parameter(names = {"--spark-master", "-ms"}, description = "Spark master", required = false) + public String sparkMaster = null; + + @Parameter(names = {"--spark-memory", "-sm"}, description = "spark memory to use", required = false) + public String sparkMemory = "1g"; + + @Parameter(names = {"--hoodie-conf"}, description = "Any configuration that can be set in the properties file " + + "(using the CLI parameter \"--props\") can also be passed command line using this parameter. This can be repeated", + splitter = IdentitySplitter.class) + public List<String> configs = new ArrayList<>(); + + @Parameter(names = {"--help", "-h"}, help = true) + public Boolean help = false; + + @Override + public String toString() { + return "HoodieTableLayoutAnalyzer {\n" + + " --base-path " + basePath + ", \n" + + " --num-days " + numDays + ", \n" + + " --start-date " + startDate + ", \n" + + " --end-date " + endDate + ", \n" + + " --enable-table-stats " + tableStats + ", \n" + + " --enable-partition-stats " + partitionStats + ", \n" + + " --output " + output + ", \n" + + " --top-n " + topN + ", \n" + + " --include-row-counts " + includeRowCounts + ", \n" + + " --analyze-table-characteristics " + analyzeTableCharacteristics + ", \n" + + " --print-top-k " + printTopK + ", \n" + + " --micro-partition-count-threshold " + microPartitionCountThreshold + ", \n" + + " --micro-partition-min-files " + microPartitionMinFiles + ", \n" + + " --micro-partition-max-avg-bytes " + microPartitionMaxAvgBytes + ", \n" + + " --micro-partition-min-age-days " + microPartitionMinAgeDays + ", \n" + + " --small-files-min-files-per-partition " + smallFilesMinFilesPerPartition + ", \n" + + " --small-files-threshold-bytes " + smallFilesThresholdBytes + ", \n" + + " --small-files-moderate-pct " + smallFilesModeratePct + ", \n" + + " --small-files-severe-pct " + smallFilesSeverePct + ", \n" + + " --small-files-min-table-commits " + smallFilesMinTableCommits + ", \n" + + " --hot-window-commits " + hotWindowCommits + ", \n" + + " --hot-partition-commit-share " + hotPartitionCommitShare + ", \n" + + " --parallelism " + parallelism + ", \n" + + " --spark-master " + sparkMaster + ", \n" + + " --spark-memory " + sparkMemory + ", \n" + + " --props " + propsFilePath + ", \n" + + " --hoodie-conf " + configs + + "\n}"; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Config config = (Config) o; + return Objects.equals(basePath, config.basePath) + && Objects.equals(numDays, config.numDays) + && Objects.equals(startDate, config.startDate) + && Objects.equals(endDate, config.endDate) + && Objects.equals(tableStats, config.tableStats) + && Objects.equals(partitionStats, config.partitionStats) + && Objects.equals(output, config.output) + && Objects.equals(topN, config.topN) + && Objects.equals(includeRowCounts, config.includeRowCounts) + && Objects.equals(analyzeTableCharacteristics, config.analyzeTableCharacteristics) + && Objects.equals(printTopK, config.printTopK) + && Objects.equals(microPartitionCountThreshold, config.microPartitionCountThreshold) + && Objects.equals(microPartitionMinFiles, config.microPartitionMinFiles) + && Objects.equals(microPartitionMaxAvgBytes, config.microPartitionMaxAvgBytes) + && Objects.equals(microPartitionMinAgeDays, config.microPartitionMinAgeDays) + && Objects.equals(smallFilesMinFilesPerPartition, config.smallFilesMinFilesPerPartition) + && Objects.equals(smallFilesThresholdBytes, config.smallFilesThresholdBytes) + && Objects.equals(smallFilesModeratePct, config.smallFilesModeratePct) + && Objects.equals(smallFilesSeverePct, config.smallFilesSeverePct) + && Objects.equals(smallFilesMinTableCommits, config.smallFilesMinTableCommits) + && Objects.equals(hotWindowCommits, config.hotWindowCommits) + && Objects.equals(hotPartitionCommitShare, config.hotPartitionCommitShare) + && Objects.equals(parallelism, config.parallelism) + && Objects.equals(sparkMaster, config.sparkMaster) + && Objects.equals(sparkMemory, config.sparkMemory) + && Objects.equals(propsFilePath, config.propsFilePath) + && Objects.equals(configs, config.configs); + } + + @Override + public int hashCode() { + return Objects.hash(basePath, numDays, startDate, endDate, tableStats, partitionStats, output, topN, includeRowCounts, + analyzeTableCharacteristics, printTopK, microPartitionCountThreshold, microPartitionMinFiles, microPartitionMaxAvgBytes, + microPartitionMinAgeDays, smallFilesMinFilesPerPartition, smallFilesThresholdBytes, smallFilesModeratePct, + smallFilesSeverePct, smallFilesMinTableCommits, hotWindowCommits, hotPartitionCommitShare, + parallelism, sparkMaster, sparkMemory, propsFilePath, configs, help); + } + } + + public static void main(String[] args) { + final Config cfg = new Config(); + JCommander cmd = new JCommander(cfg, null, args); + + if (cfg.help || args.length == 0) { + cmd.usage(); + System.exit(1); + } + + Map<String, String> sparkConfigMap = new HashMap<>(); + sparkConfigMap.put("spark.executor.memory", cfg.sparkMemory); + JavaSparkContext jsc = UtilHelpers.buildSparkContext("Table-Layout-Analyzer", cfg.sparkMaster, false, sparkConfigMap); + + try { + HoodieTableLayoutAnalyzer analyzer = new HoodieTableLayoutAnalyzer(jsc, cfg); + analyzer.run(); + } catch (TableNotFoundException e) { + log.warn("The Hudi data table is not found: [{}].", cfg.basePath, e); + } catch (Throwable throwable) { + log.error("Failed to analyze table layout for {}", cfg, throwable); + } finally { + jsc.stop(); + } + } + + public void run() { + try { + log.info(cfg.toString()); + log.info(" ****** Analyzing table layout ******"); + + // Determine starting and ending date intervals for filtering data files. + LocalDate[] dateInterval = getUserSpecifiedDateInterval(cfg); + + if (cfg.propsFilePath != null) { + List<String> filePaths = getFilePaths(cfg.propsFilePath, jsc.hadoopConfiguration()); + for (String filePath : filePaths) { + analyzeTable(filePath, dateInterval); + } + } else { + if (cfg.basePath == null) { + throw new HoodieIOException("Base path needs to be set."); + } + analyzeTable(cfg.basePath, dateInterval); + } + + } catch (Exception e) { + throw new HoodieException("Unable to analyze table layout for " + cfg.basePath, e); + } + } + + private void analyzeTable(String basePath, LocalDate[] dateInterval) throws IOException { + log.info("Processing table {}", basePath); + StorageConfiguration<?> storageConf = HadoopFSUtils.getStorageConfWithCopy(jsc.hadoopConfiguration()); + HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder() + .setBasePath(basePath) + .setConf(storageConf.newInstance()).build(); + HoodieTableConfig tableConfig = metaClient.getTableConfig(); + boolean mdtEnabled = tableConfig.isMetadataPartitionAvailable(MetadataPartitionType.FILES); + boolean colStatsAvailable = tableConfig.isMetadataPartitionAvailable(MetadataPartitionType.COLUMN_STATS); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder() + .enable(mdtEnabled) + .build(); + HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); + + // Both tableMetadata and fileSystemView are AutoCloseable. Under --props-path batch mode + // (multiple tables per process) leaving them open leaks metadata readers and file-listing + // handles per table, which can hit FD / heap limits over long batches. + // + // The file system view is built once per table and honors the table's metadata-table + // setting, so on large tables partitions are listed from the metadata table instead of + // one storage listing per partition. + try (HoodieTableMetadata tableMetadata = metaClient.getTableFormat().getMetadataFactory() + .create(engineContext, metaClient.getStorage(), metadataConfig, basePath); + HoodieTableFileSystemView fileSystemView = FileSystemViewManager + .createInMemoryFileSystemView(new HoodieLocalEngineContext(storageConf), metaClient, metadataConfig)) { + + List<String> allPartitions = tableMetadata.getAllPartitionPaths(); + + // As a sanity check, throw exception and exit early if date interval is specified, but the first partition does not have + // date. + if (dateInterval != null && !allPartitions.isEmpty() && getPartitionDate(allPartitions.get(0)) == null) { + throw new HoodieException( + "Cannot apply --start-date, --end-date, or --num-days when partition does not contain date. Interval: " + Arrays.toString(dateInterval) + ", Partition Name: " + allPartitions.get(0)); + } + + List<PartitionRow> rows = new ArrayList<>(); + // Table-level reservoirs are one per run, so 1M slots (~8 MB each) is fine and gives + // accurate p95/p99 on tables with hundreds of thousands of files. + final Histogram tableSizeHistogram = new Histogram(new UniformReservoir(1_000_000)); + final Histogram tableFileCountHistogram = new Histogram(new UniformReservoir(1_000_000)); + for (String partition : allPartitions) { + if (!isPartitionInRange(partition, dateInterval)) { + continue; + } + PartitionRow row = collectPartitionRow(partition, fileSystemView, tableMetadata, colStatsAvailable, storageConf); + row.sizes.forEach(tableSizeHistogram::update); + tableFileCountHistogram.update(row.fileCount); + rows.add(row); + } + + // Sort partitions by total bytes descending so output (and --top-n) shows the largest first. + rows.sort(Comparator.comparingLong((PartitionRow r) -> r.totalBytes).reversed()); + + TableCharacteristics characteristics = cfg.analyzeTableCharacteristics + ? analyzeCharacteristics(metaClient, rows) + : null; + + OutputFormat output = parseOutputFormat(cfg.output); + if (output == OutputFormat.JSON) { + emitJson(basePath, rows, tableSizeHistogram, tableFileCountHistogram, mdtEnabled, characteristics); + } else { + emitTable(basePath, rows, tableSizeHistogram, tableFileCountHistogram, mdtEnabled, characteristics); + } + } catch (HoodieException he) { + throw he; + } catch (Exception e) { + throw new HoodieException("Failure while analyzing table layout for " + basePath, e); + } + } + + /** + * Applies the optional date interval to a partition name. A partition is in range when: + * 1. no interval is specified, or the partition name does not contain a date; + * 2. only a start date is specified and the partition date is on or after it; + * 3. only an end date is specified and the partition date is before it; + * 4. both are specified and the partition date lies in [startDate, endDate). + */ + private static boolean isPartitionInRange(String partition, LocalDate[] dateInterval) { + if (dateInterval == null) { + return true; + } + LocalDate partitionDate = getPartitionDate(partition); + if (partitionDate == null) { + return true; + } + LocalDate startDate = dateInterval[0]; + LocalDate endDate = dateInterval[1]; + boolean afterStart = startDate == null || !partitionDate.isBefore(startDate); + boolean beforeEnd = endDate == null || partitionDate.isBefore(endDate); + return afterStart && beforeEnd; + } + + private PartitionRow collectPartitionRow(String partition, HoodieTableFileSystemView fileSystemView, + HoodieTableMetadata tableMetadata, boolean colStatsAvailable, + StorageConfiguration<?> storageConf) { + List<HoodieBaseFile> baseFiles = fileSystemView.getLatestBaseFiles(partition).collect(Collectors.toList()); + + // The per-partition histogram is only rendered under --enable-partition-stats. Skip the + // allocation when it isn't needed: UniformReservoir eagerly allocates an AtomicLongArray + // of its capacity (8 bytes per slot) on construction, so a 1M-slot reservoir is ~8 MB + // per partition and would explode driver heap on tables with many partitions. When we + // do build it, use a 4096-slot reservoir, plenty of headroom for the per-partition + // file count in practice, while keeping the per-partition cost at ~32 KB. + Histogram partitionSizeHist = cfg.partitionStats ? new Histogram(new UniformReservoir(4096)) : null; + long partitionTotalBytes = 0L; + List<Long> sizes = new ArrayList<>(baseFiles.size()); + for (HoodieBaseFile baseFile : baseFiles) { + long size = baseFile.getFileSize(); + if (partitionSizeHist != null) { + partitionSizeHist.update(size); + } + sizes.add(size); + partitionTotalBytes += size; + } + + Long partitionRowCount = null; + if (cfg.includeRowCounts && !baseFiles.isEmpty()) { + partitionRowCount = computeRowCount(partition, baseFiles, tableMetadata, colStatsAvailable, storageConf); + } + return new PartitionRow(partition, baseFiles.size(), partitionTotalBytes, sizes, partitionSizeHist, partitionRowCount); + } + + // ---- table characteristics (micro / small / hot detectors) ---------------- + + /** + * Runs the three table-characteristic detectors over the per-partition row set. Each + * detector's verdict is independent; the umbrella flag {@code --analyze-table-characteristics} + * gates whether any of them run, and thresholds are individually configurable. + */ + private TableCharacteristics analyzeCharacteristics(HoodieTableMetaClient metaClient, List<PartitionRow> rows) { + TableCharacteristics tc = new TableCharacteristics(); + tc.tableType = metaClient.getTableType(); + tc.tableAgeDays = computeTableAgeDays(metaClient); + tc.numPartitions = rows.size(); + + runMicroPartitionDetector(rows, tc); + runSmallFileDetector(metaClient, rows, tc); + + try { + computeHotPartitions(metaClient, tc); + } catch (Exception e) { + log.warn("Hot-partition detection failed: {}", e.getMessage()); + tc.hotPartitionDetectionFailed = true; + } + return tc; + } + + /** + * Micro-partition verdict: table is "micro" when partition count exceeds the count threshold, + * OR (on mature tables) any partition has many small files. The size rule is gated by + * table age, since new tables legitimately have small partitions. + */ + private void runMicroPartitionDetector(List<PartitionRow> rows, TableCharacteristics tc) { + boolean countTrigger = tc.numPartitions > cfg.microPartitionCountThreshold; Review Comment: Confirmed and fixed in `85479cfb94a2`. You are right that the date filter was changing the question rather than bounding the work: `tc.numPartitions` came from `rows`, which is post-filter, so `--num-days 7` against a 50K-partition table judged the count rule on about seven partitions. The count rule now reads `allPartitions.size()` — unfiltered — because "does this table have too many partitions?" is a question about the table, not about the window being read. I also added `partitionsExamined` next to `numPartitions` in the JSON, so a date-scoped run says plainly how much of the table the *other* detectors looked at. That felt better than your alternative of noting the scope in the summary text, since it is a number the agent can reason about rather than prose it has to parse. The hot-partition scan ignoring the date filter I left alone: it scans the last N *commits* rather than partitions, so a partition-date window is not really meaningful for it, and it is now informational anyway (see the `overallStatus` thread). -- 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]
