nsivabalan commented on code in PR #20152: URL: https://github.com/apache/hudi/pull/20152#discussion_r4284077095
########## 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; + boolean sizeTriggerEligible = tc.tableAgeDays >= cfg.microPartitionMinAgeDays; + int sizeMatches = 0; + if (sizeTriggerEligible) { + for (PartitionRow r : rows) { + if (r.fileCount >= cfg.microPartitionMinFiles && r.avgFileSize() < cfg.microPartitionMaxAvgBytes) { + sizeMatches++; + } + } + } + tc.microCountTrigger = countTrigger; + tc.microSizeTrigger = sizeMatches > 0; + tc.microSizeMatchCount = sizeMatches; + tc.microPartitioned = countTrigger || tc.microSizeTrigger; + // Top-K evidence: partitions with the smallest avg file size among those satisfying the + // micro file-count gate. Independent of the verdict; useful even when the verdict is "no". + if (cfg.printTopK > 0) { + tc.microTopKSmallest.addAll(collectTopKSmallest(rows, cfg.microPartitionMinFiles, cfg.printTopK)); + } + } + + /** + * Small-file pile-up verdict (CLEAN / MODERATE / SEVERE / SKIPPED) based on per-partition + * prevalence. Count partitions with >= minFiles as "qualifying"; among those, count partitions + * with avgFileSize < threshold as "flagged". Verdict thresholds are flagged/qualifying ratio + * against the moderate/severe pct gates. Skipped when the table has too few commits, since + * the signal is unreliable on freshly-ingested tables. + */ + private void runSmallFileDetector(HoodieTableMetaClient metaClient, List<PartitionRow> rows, TableCharacteristics tc) { + long totalIngestCommits = countTotalIngestCommits(metaClient); + tc.smallFileTableIngestCommitCount = totalIngestCommits; + if (totalIngestCommits < cfg.smallFilesMinTableCommits) { + tc.smallFileVerdict = SmallFileVerdict.SKIPPED; + return; + } + int qualifying = 0; + int flagged = 0; + for (PartitionRow r : rows) { + if (r.fileCount < cfg.smallFilesMinFilesPerPartition) { + continue; + } + qualifying++; + if (r.avgFileSize() < cfg.smallFilesThresholdBytes) { + flagged++; + } + } + tc.smallFileQualifyingPartitions = qualifying; + tc.smallFileFlaggedPartitions = flagged; + tc.smallFileFlaggedPct = qualifying == 0 ? 0.0 : (double) flagged / qualifying; + tc.smallFileVerdict = classifySmallFileVerdict(qualifying, tc.smallFileFlaggedPct); + // Top-K evidence: partitions with the smallest avg file size among the qualifying set. + if (cfg.printTopK > 0) { + tc.smallFileTopKSmallest.addAll(collectTopKSmallest(rows, cfg.smallFilesMinFilesPerPartition, cfg.printTopK)); + } + } + + /** + * Returns the top-K partitions with the smallest avg file size among those satisfying the + * given file-count gate, in ascending avg-file-size order. + */ + private static List<SmallestPartition> collectTopKSmallest(List<PartitionRow> rows, int minFiles, int k) { + List<SmallestPartition> qualifying = new ArrayList<>(); + for (PartitionRow r : rows) { + if (r.fileCount < minFiles) { + continue; + } + qualifying.add(new SmallestPartition(r.partition, r.fileCount, r.totalBytes, r.avgFileSize())); + } + qualifying.sort(Comparator.comparingLong((SmallestPartition p) -> p.avgFileSize) + .thenComparingLong(p -> p.totalBytes)); + return qualifying.subList(0, Math.min(k, qualifying.size())); + } + + private SmallFileVerdict classifySmallFileVerdict(int qualifying, double flaggedPct) { + if (qualifying == 0) { + return SmallFileVerdict.CLEAN; + } else if (flaggedPct >= cfg.smallFilesSeverePct) { + return SmallFileVerdict.SEVERE; + } else if (flaggedPct >= cfg.smallFilesModeratePct) { + return SmallFileVerdict.MODERATE; + } + return SmallFileVerdict.CLEAN; + } + + /** + * Estimates table age in days as the earlier of two cheap signals: the modification time of + * {@code hoodie.properties} and the first instant on the active timeline. Neither is exact + * on its own. The properties file is rewritten on upgrades and when metadata partitions + * change, so its mtime can be later than table creation; the active timeline's first + * instant moves forward as commits are archived. Taking the earlier of the two never + * over-estimates age, and is only an under-estimate when both signals are later than the + * true creation time. The age gates the size-based micro-partition rule, so an + * under-estimate errs on the side of not flagging. + */ + private long computeTableAgeDays(HoodieTableMetaClient metaClient) { + LocalDate today = LocalDate.now(); + long ageDays = 0; + try { + StoragePath propsPath = new StoragePath(metaClient.getMetaPath(), HoodieTableConfig.HOODIE_PROPERTIES_FILE); + long mtime = metaClient.getStorage().getPathInfo(propsPath).getModificationTime(); + if (mtime > 0) { + LocalDate modified = Instant.ofEpochMilli(mtime).atZone(ZoneId.systemDefault()).toLocalDate(); + ageDays = Math.max(ageDays, ChronoUnit.DAYS.between(modified, today)); + } + } catch (Exception e) { + log.warn("Failed to read hoodie.properties modification time for table age: {}", e.getMessage()); + } + + try { + Option<HoodieInstant> firstOpt = metaClient.getActiveTimeline().firstInstant(); + if (firstOpt.isPresent()) { + String ts = firstOpt.get().requestedTime(); + // Hudi instant times are yyyyMMddHHmmssSSS. Parse the date portion. + if (ts.length() >= 8) { + LocalDate first = LocalDate.parse(ts.substring(0, 8), DateTimeFormatter.ofPattern("yyyyMMdd")); + ageDays = Math.max(ageDays, ChronoUnit.DAYS.between(first, today)); + } + } + } catch (Exception e) { + log.warn("Failed to derive table age from the active timeline: {}", e.getMessage()); + } + return ageDays; + } + + /** + * Counts total completed ingest commits (commit + deltacommit + replacecommit) across + * active and archived timelines. Used to gate the small-file verdict: tables with too + * few commits aren't worth scoring because the signal is dominated by initial-write + * noise. The archived timeline is only consulted when the active timeline alone is under + * the gate, to avoid the archive scan on mature tables. + * + * <p>Replacecommits from clustering are excluded, since they aren't ingests and would + * inflate the count on tables with frequent clustering. INSERT_OVERWRITE / + * INSERT_OVERWRITE_TABLE replacecommits are real ingests and are kept. Distinguishing the + * two requires reading each replacecommit's metadata, which is only done for the active + * timeline: the archived timeline is loaded without instant details, so its replacecommits + * are skipped and the count is a lower bound there. + */ + private long countTotalIngestCommits(HoodieTableMetaClient metaClient) { + try { + long activeCount = countIngestCommits(metaClient.getActiveTimeline(), true); + if (activeCount >= cfg.smallFilesMinTableCommits) { + return activeCount; + } + // Need archived to see if the table has actually received enough commits. + long archivedCount = countIngestCommits(metaClient.getArchivedTimeline(), false); + return activeCount + archivedCount; + } catch (Exception e) { + log.warn("Failed to count ingest commits: {}", e.getMessage()); + return 0; + } + } + + /** + * Counts completed ingest commits on a single timeline. Commit/deltacommit are counted + * unconditionally; each replacecommit is read to distinguish real ingests + * (INSERT_OVERWRITE / INSERT_OVERWRITE_TABLE) from clustering, which is skipped. When + * {@code detailsReadable} is false, replacecommits are skipped rather than read. + */ + private long countIngestCommits(HoodieTimeline timeline, boolean detailsReadable) { + return timeline.filterCompletedInstants().getInstantsAsStream() + .filter(i -> { + String action = i.getAction(); + if (action.equals(HoodieTimeline.COMMIT_ACTION) || action.equals(HoodieTimeline.DELTA_COMMIT_ACTION)) { + return true; + } + if (!action.equals(HoodieTimeline.REPLACE_COMMIT_ACTION) || !detailsReadable) { + return false; + } + try { + return timeline.readCommitMetadata(i).getOperationType() != WriteOperationType.CLUSTER; + } catch (Exception e) { + // Best-effort: an unreadable replacecommit body shouldn't fail the whole run. + // Skip it; undercounting is preferable to misclassifying. + log.warn("Skipping replacecommit {} during ingest-commit count: {}", i.requestedTime(), e.toString()); + return false; + } + }).count(); + } + + /** + * Scans the last N completed ingest commits (commit + deltacommit + replacecommit, excluding + * compaction and clustering operations) and accumulates per-partition commit-frequency / bytes / + * records, then flags partitions whose participation share exceeds the threshold. + * Replacecommit is kept because INSERT_OVERWRITE/INSERT_OVERWRITE_TABLE are real ingests; + * clustering also writes replacecommit on older table versions but is filtered out by the + * operation-type check. + * + * <p>Known limitation: write stats for log files (Merge-on-Read delta commits) are noisier + * than base-file write stats, so the resulting "hot" scoring favors partitions that get + * rewritten frequently over partitions that are appended to. For log-heavy Merge-on-Read + * workloads the hot-partition signal may under-report. + */ + private void computeHotPartitions(HoodieTableMetaClient metaClient, TableCharacteristics tc) { + HoodieActiveTimeline timeline = metaClient.reloadActiveTimeline(); + List<HoodieInstant> instants = candidateIngestInstantsNewestFirst(timeline); + + Map<String, HotPartitionCounters> counters = new LinkedHashMap<>(); + int kept = aggregateHotPartitionCounters(timeline, instants, counters); + tc.hotWindowEffectiveCommits = kept; + tc.hotPartitions.addAll(flagHotPartitions(counters, kept)); + } + + /** + * Most-recent first. We walk until we've accumulated {@code hotWindowCommits} post-filter + * ingest commits (true ingests, not COMPACT/CLUSTER). The window can grow to fill + * arbitrarily many raw instants on tables where compaction/clustering dominate recent + * activity, bounded only by the active timeline size. + */ + private static List<HoodieInstant> candidateIngestInstantsNewestFirst(HoodieActiveTimeline timeline) { + return timeline.getCommitsTimeline().filterCompletedInstants() + .getInstantsAsStream() + .filter(i -> { + String action = i.getAction(); + return action.equals(HoodieTimeline.COMMIT_ACTION) + || action.equals(HoodieTimeline.DELTA_COMMIT_ACTION) + || action.equals(HoodieTimeline.REPLACE_COMMIT_ACTION); + }) + .sorted(Comparator.comparing(HoodieInstant::requestedTime).reversed()) + .collect(Collectors.toList()); + } + + /** + * Walks the candidate instants and accumulates per-partition counters until we have + * {@code hotWindowCommits} true ingest commits. Returns the actual number kept (the + * effective window size). + */ + private int aggregateHotPartitionCounters(HoodieActiveTimeline timeline, List<HoodieInstant> instants, + Map<String, HotPartitionCounters> counters) { + int kept = 0; + for (HoodieInstant inst : instants) { + if (kept >= cfg.hotWindowCommits) { + break; + } + try { + HoodieCommitMetadata cm = timeline.readCommitMetadata(inst); + WriteOperationType op = cm.getOperationType(); + // Exclude compaction + clustering: they touch many partitions per commit and + // would skew hotness toward "every partition is hot". + if (op == WriteOperationType.COMPACT || op == WriteOperationType.CLUSTER) { + continue; + } + kept++; + Map<String, List<HoodieWriteStat>> partitionToWriteStats = cm.getPartitionToWriteStats(); + if (partitionToWriteStats == null) { + continue; + } + for (Map.Entry<String, List<HoodieWriteStat>> e : partitionToWriteStats.entrySet()) { + counters.computeIfAbsent(e.getKey(), k -> new HotPartitionCounters()).add(e.getValue()); + } + } catch (Exception e) { + // Best-effort: an empty/corrupt instant body shouldn't fail the whole run. Log at WARN + // so the operator sees which instants were skipped. + log.warn("Skipping instant {} during hot-partition scan: {}", inst.requestedTime(), e.toString()); + } + } + return kept; + } + + /** + * Returns partitions that appear in at least {@code hotPartitionCommitShare} of the + * effective window, sorted by commitCount desc, bytesWritten desc. + */ + private List<HotPartition> flagHotPartitions(Map<String, HotPartitionCounters> counters, int kept) { + List<HotPartition> hot = new ArrayList<>(); + // When no ingest commits fell in the window (e.g., recent activity was all COMPACT/CLUSTER), + // ceil(share * 0) is 0 and every partition would falsely be flagged as hot. Bail early. + if (kept == 0) { + return hot; + } + long minCommitCount = hotPartitionMinCommitCount(kept); + for (Map.Entry<String, HotPartitionCounters> e : counters.entrySet()) { + HotPartitionCounters c = e.getValue(); + if (c.commitCount >= minCommitCount) { + hot.add(new HotPartition(e.getKey(), c.commitCount, c.bytesWritten, c.recordsWritten)); + } + } + hot.sort(Comparator.comparingLong((HotPartition h) -> h.commitCount).reversed() + .thenComparing(Comparator.comparingLong((HotPartition h) -> h.bytesWritten).reversed())); + return hot; + } + + private long hotPartitionMinCommitCount(int effectiveWindow) { + return (long) Math.ceil(cfg.hotPartitionCommitShare * effectiveWindow); + } + + /** + * Computes the total record count for one partition. Prefers metadata-table column stats + * when available (one lookup per partition), falls back to Parquet footer reads per file. + * Returns null on any error so callers can render "-" rather than failing the run. + * + * <p>Partial-fallback behavior: if column stats return data for some files but not others + * (common right after a write before the index catches up), the successful lookups are + * kept and only the misses fall back to footer reads. + */ + private Long computeRowCount(String partition, List<HoodieBaseFile> baseFiles, HoodieTableMetadata tableMetadata, + boolean colStatsAvailable, StorageConfiguration<?> storageConf) { + long total = 0L; + List<HoodieBaseFile> filesNeedingFooterRead; + + if (colStatsAvailable) { + filesNeedingFooterRead = new ArrayList<>(); + try { + List<Pair<String, String>> partitionFilePairs = new ArrayList<>(); + for (HoodieBaseFile bf : baseFiles) { + partitionFilePairs.add(Pair.of(partition, bf.getFileName())); + } + Map<Pair<String, String>, HoodieMetadataColumnStats> stats = tableMetadata.getColumnStats(partitionFilePairs, ROW_COUNT_COLUMN); + for (HoodieBaseFile bf : baseFiles) { + HoodieMetadataColumnStats cs = stats.get(Pair.of(partition, bf.getFileName())); + if (cs == null || cs.getValueCount() == null) { + filesNeedingFooterRead.add(bf); + } else { + total += cs.getValueCount(); + } + } + } catch (Exception e) { + log.warn("Metadata column-stats lookup failed for partition {}, falling back to footer reads: {}", partition, e.getMessage()); + // The column-stats batch failed entirely; treat every file as missing. + total = 0L; + filesNeedingFooterRead = new ArrayList<>(baseFiles); + } + } else { + filesNeedingFooterRead = baseFiles; + } + + Configuration hadoopConf = storageConf.unwrapAs(Configuration.class); + for (HoodieBaseFile bf : filesNeedingFooterRead) { + try { + total += readParquetRowCount(bf.getPath(), hadoopConf); + } catch (Exception e) { + log.warn("Failed to read row count from {}: {}", bf.getPath(), e.getMessage()); + return null; + } + } + return total; + } + + /** + * Reads the record count from a Parquet file's footer. Only the footer is read, not the row + * groups themselves. + */ + private static long readParquetRowCount(String pathStr, Configuration hadoopConf) throws IOException { + try (ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(new Path(pathStr), hadoopConf))) { + return reader.getRecordCount(); + } + } + + // ---- output --------------------------------------------------------------- + + /** + * Sums per-partition record counts; null when row counts were not requested, or when any + * partition's count is unknown (so a partial sum is never reported as a table total). + */ + private Long tableTotalRows(List<PartitionRow> rows) { + if (!cfg.includeRowCounts || rows.isEmpty()) { + return null; + } + long sum = 0L; + for (PartitionRow r : rows) { + if (r.numRecords == null) { + return null; + } + sum += r.numRecords; + } + return sum; + } + + private List<PartitionRow> visibleRows(List<PartitionRow> rows) { + return cfg.topN > 0 && rows.size() > cfg.topN ? rows.subList(0, cfg.topN) : rows; + } + + private void emitTable(String basePath, List<PartitionRow> rows, + Histogram tableSizeHist, Histogram tableFileCountHist, + boolean mdtEnabled, TableCharacteristics characteristics) { + long tableTotalBytes = rows.stream().mapToLong(r -> r.totalBytes).sum(); + long tableTotalFiles = rows.stream().mapToLong(r -> r.fileCount).sum(); + Long tableTotalRows = tableTotalRows(rows); + + System.out.println("Table: " + basePath + " (MDT=" + (mdtEnabled ? "on" : "off") + ")"); + System.out.println(); + + if (cfg.partitionStats && !rows.isEmpty()) { + emitPartitionTable(rows); + } + + if (cfg.tableStats || !cfg.partitionStats) { + Snapshot ts = tableSizeHist.getSnapshot(); + System.out.println("Table-level file size distribution:"); + System.out.printf(" numFiles=%d totalBytes=%s%n", ts.size(), formatBytes(tableTotalBytes)); Review Comment: Confirmed and fixed in `1ea199b3d6b4`. `numFiles` now comes from `tableTotalFiles`, which is already summed exactly from the per-partition rows — the reservoir's `size()` saturates at its capacity, so a table with more than a million base files would have reported exactly a million. The per-partition reservoir is also sized to the partition now (`min(4096, baseFiles.size())`) rather than always allocating the cap — see the thread on line 530. I left the table-level min/max/mean sampled. For a table-wide distribution over a million-slot reservoir the sampling error is not worth the memory of an exact copy, and `PartitionRow.sizes` is per-partition rather than table-wide, so using it here would mean materialising every file size on the driver. ########## 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) { Review Comment: Both confirmed and fixed in `1ea199b3d6b4`. The `TableNotFoundException` branch was indeed dead — `run()` wraps everything in `HoodieException` at `:423` before `main`'s catch at `:393` could match. The cause is now unwrapped so a missing table still gets its own message rather than the generic one. And you are right about the exit code mattering more here than usual: this tool is driven by agents and schedulers, so a failed run exiting 0 is indistinguishable from a run that found nothing. It now exits 1 after `jsc.stop()`. -- 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]
