This is an automated email from the ASF dual-hosted git repository.
nsivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new a974f447089e feat(utilities): add record-level index sizing check to
HoodieTableHealthChecker (#20153)
a974f447089e is described below
commit a974f447089eaea3d9366ee495e88d5b5a349d58
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Thu Oct 22 08:16:57 2026 -0700
feat(utilities): add record-level index sizing check to
HoodieTableHealthChecker (#20153)
* feat(utilities): add record-level index sizing check to
HoodieTableHealthChecker
Adds RecordIndexSizingCheck (--checks record-index). The record index's
file-group count is
fixed at initialization, so the check re-runs the writer's own sizing
estimate against the index
as it is now and compares the result with the file groups it has,
globally or per data partition
for a partitioned index. It also flags any file group past the configured
maximum size.
The record estimate is taken from base-file bytes, falling back to log
bytes before the first
metadata compaction. A floor-bound ideal (min == max, or an index below
one file group's worth of
records) is reported as such and never called undersized. The worst data
partition is the one
with a real shortfall, not the lowest ratio. The report echoes the
effective sizing, the
estimate basis and the ideal's source so a verdict can be audited.
The remedy is a re-bootstrap with corrected sizing; for a partitioned
index the check says why
HoodieIndexer cannot do that today (#20343) rather than naming a command
that would build the
wrong index. Part of #19262.
---
.../hudi/utilities/HoodieTableHealthChecker.java | 6 +-
.../main/java/org/apache/hudi/utilities/README.md | 44 ++
.../utilities/health/RecordIndexSizingCheck.java | 595 +++++++++++++++++++++
.../utilities/TestHoodieTableHealthChecker.java | 2 +-
.../health/TestRecordIndexSizingCheck.java | 298 +++++++++++
.../TestRecordIndexSizingCheckWithSpark.java | 428 +++++++++++++++
6 files changed, 1370 insertions(+), 3 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
index 2d6625a008b3..8a910f1a44e1 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
@@ -32,6 +32,7 @@ import org.apache.hudi.utilities.health.HealthCheckResult;
import org.apache.hudi.utilities.health.HealthStatus;
import org.apache.hudi.utilities.health.MetadataCompactionLagCheck;
import org.apache.hudi.utilities.health.MetadataTableSyncCheck;
+import org.apache.hudi.utilities.health.RecordIndexSizingCheck;
import org.apache.hudi.utilities.health.SavepointArchivalBlockCheck;
import com.beust.jcommander.JCommander;
@@ -150,7 +151,8 @@ public class HoodieTableHealthChecker implements
Serializable {
new SavepointArchivalBlockCheck(),
new ArchivalCadenceCheck(),
new MetadataTableSyncCheck(),
- new MetadataCompactionLagCheck());
+ new MetadataCompactionLagCheck(),
+ new RecordIndexSizingCheck());
}
/**
@@ -291,7 +293,7 @@ public class HoodieTableHealthChecker implements
Serializable {
public String basePath = null;
@Parameter(names = {"--checks"}, description = "Comma-separated checks to
run, or 'all'. Available: "
- + "compaction, cleaner, savepoint, archival, mdt-sync, mdt-compaction")
+ + "compaction, cleaner, savepoint, archival, mdt-sync, mdt-compaction,
record-index")
public String checks = "all";
@Parameter(names = {"--apply-all-defaults"}, description = "Evaluate
checks against Hudi default values for any "
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
index f00d6072249d..73670cb5d49b 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
@@ -200,6 +200,50 @@ rebuild the metadata table.
Governing properties: `hoodie.metadata.enable`,
`hoodie.metadata.compact.max.delta.commits`.
+### `record-index`
+
+Whether the record-level index has enough file groups for the table it serves.
Skipped when the
+table's metadata table has no `record_index` partition.
+
+The file-group count of the record index is decided once, at initialization,
from the record count
+at that moment and the writer's sizing configuration, and it never changes
afterwards. A table that
+has since grown well past that point has index file groups far larger than
intended: every index
+lookup reads a larger file, and every index write concentrates on the same few
file groups.
+
+The check re-runs the writer's own sizing estimate against the table as it is
now and compares the
+result against the file-group count the index actually has. The record count
is *estimated* from
+the index's own footprint on storage divided by the same average record size
the writer sizes with,
+so no data file is scanned; index files are compressed, so the estimate runs
low. Two signals are
+reported:
+
+- **Undersized.** The actual count is below half the ideal. The ideal already
carries the writer's
+ growth factor, so an index initialized on a smaller table routinely sits
somewhat below it; the
+ 0.5 ratio keeps ordinary growth from alerting while a table that has doubled
past its sizing
+ still does.
+- **Oversized file group.** The largest file group is past the configured
maximum size with 1.5x
+ slack. Log files push a group past the maximum transiently between metadata
compactions, so
+ alerting at exactly the maximum would fire on healthy tables.
+
+A partitioned record index is sized per data partition, so for that layout the
count comparison is
+made per data partition and the worst one is reported
(`observed.worst.data.partition`). The layout
+the writer sizes for is read from its configuration and the layout on storage
from the index's
+file-group names; if they disagree, the count comparison is skipped,
`effective.count.comparison`
+says why, and only the file-group size rule runs.
+
+Both share one remedy, and it is disruptive: the index must be
re-bootstrapped. Drop it with
+`metadata delete-record-index` in `hudi-cli` (or `HoodieIndexer --mode
dropindex --index-types
+RECORD_INDEX`), correct the sizing configuration, then rebuild with
`HoodieIndexer --mode
+scheduleAndExecute --index-types RECORD_INDEX`. The record index is
unavailable until the rebuild
+completes and the drop cannot be undone in place, so plan it as a maintenance
window.
+
+Governing properties:
`hoodie.metadata.global.record.level.index.min.filegroup.count`,
+`hoodie.metadata.global.record.level.index.max.filegroup.count`,
+`hoodie.metadata.record.index.max.filegroup.size`,
`hoodie.metadata.record.index.growth.factor`.
+When `hoodie.metadata.record.level.index.enable` is `true` the partitioned keys
+`hoodie.metadata.record.level.index.min.filegroup.count` and
+`hoodie.metadata.record.level.index.max.filegroup.count` govern instead. The
pre-rename spellings
+of the count keys are accepted.
+
## Exit codes
| Code | Meaning |
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/RecordIndexSizingCheck.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/RecordIndexSizingCheck.java
new file mode 100644
index 000000000000..9028aeb1cd40
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/RecordIndexSizingCheck.java
@@ -0,0 +1,595 @@
+/*
+ * 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.health;
+
+import org.apache.hudi.common.config.ConfigProperty;
+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.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.core.index.record.HoodieRecordIndex;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.metadata.MetadataPartitionType;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_RECORD_INDEX;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.RECORD_INDEX_AVERAGE_RECORD_SIZE;
+
+/**
+ * Checks whether the record-level index is sized appropriately for the table
it serves.
+ *
+ * <p>The number of file groups in the record index is decided once, when the
index is initialized,
+ * from the record count at that moment and the writer's sizing configuration.
It never changes
+ * afterwards: every subsequent index write lands in one of those same file
groups. A table that has
+ * grown well past what its index was sized for therefore has file groups that
are each far larger
+ * than the writer's configuration intended, which slows every index lookup
and concentrates every
+ * index write on the same few file groups.
+ *
+ * <p>So the check re-runs the writer's own sizing estimate against the table
as it is now and
+ * compares that ideal against the file-group count the index actually has. It
also looks at the
+ * largest file group directly, since a group past the configured maximum size
is the symptom the
+ * operator will feel regardless of how the count came out. Both signals share
one remedy: the index
+ * has to be re-bootstrapped with corrected sizing, which is disruptive and is
spelled out in the
+ * findings.
+ *
+ * <p>A partitioned record index is sized per data partition, not as a whole,
so for that layout
+ * the count comparison is made per data partition and the worst one is
reported. Which layout the
+ * writer sizes for is read from its configuration and which layout is on
storage is read from the
+ * index's file-group names; if the two disagree, the count comparison is
skipped and the report says
+ * so, while the file-group size rule still runs.
+ *
+ * <p>The record count is estimated from the index's own footprint on storage
divided by the same
+ * average record size the writer sizes with, so that no data file is scanned.
The estimate is
+ * conservative -- index files are compressed, so the true count is higher --
and it is labelled as
+ * an estimate in the report.
+ */
+public class RecordIndexSizingCheck implements HealthCheck {
+
+ static final String NAME = "record-index";
+
+ /**
+ * Actual-to-ideal file-group ratio below which the index is called
undersized. The ideal count
+ * already carries the writer's growth factor, so an index initialized on a
smaller table sits
+ * somewhat below 1.0 during ordinary growth. The slack keeps that from
alerting while a table
+ * that has grown past what its index was sized for still does.
+ */
+ static final double MIN_ACCEPTABLE_FILE_GROUP_RATIO = 0.5;
+
+ /**
+ * How far past the configured maximum a single file group may grow before
the index is called
+ * unhealthy. File groups are bounded only at initialization; between
metadata compactions, log
+ * files push a group past the configured size transiently, so alerting at
exactly the maximum
+ * would fire on healthy tables.
+ */
+ static final double MAX_FILE_GROUP_SIZE_SLACK_FACTOR = 1.5;
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Whether the record-level index has enough file groups for the
table it serves";
+ }
+
+ @Override
+ public boolean appliesTo(HealthCheckContext context) {
+ return
context.getMetaClient().getTableConfig().isMetadataPartitionAvailable(MetadataPartitionType.RECORD_INDEX);
+ }
+
+ @Override
+ public HealthCheckResult check(HealthCheckContext context) throws Exception {
+ if (!appliesTo(context)) {
+ return HealthCheckResult.skipped(NAME, String.format(
+ "Table has no record-level index: '%s' in the table configuration
does not list '%s'",
+ HoodieTableConfig.TABLE_METADATA_PARTITIONS.key(),
MetadataPartitionType.RECORD_INDEX.getPartitionPath()));
+ }
+
+ TypedProperties props = context.getProps();
+ // The layout is a fact about the table, so read it from the index
definition on storage rather
+ // than from a writer property. The property describes what a writer
wants, defaults to false
+ // when unset, and would silently send a partitioned table down the global
branch -- picking the
+ // wrong sizing keys, naming the wrong keys in the skip message, and
skipping the per-data-
+ // partition comparison entirely under --apply-all-defaults.
+ Option<Boolean> layoutOnStorage =
partitionedIndexOnStorage(context.getMetaClient());
+ boolean partitionedIndex = layoutOnStorage.orElseGet(
+ () -> ConfigUtils.getBooleanWithAltKeys(props,
HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP));
+ ConfigProperty<Integer> minCountProp = partitionedIndex
+ ? HoodieMetadataConfig.RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP
+ :
HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP;
+ ConfigProperty<Integer> maxCountProp = partitionedIndex
+ ? HoodieMetadataConfig.RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP
+ :
HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP;
+
+ Option<String> missing = firstMissingConfig(context, minCountProp,
maxCountProp,
+ HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP,
+ HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP);
+ if (missing.isPresent()) {
+ return context.skipForMissingConfig(NAME, missing.get());
+ }
+
+ SizingConfig sizing = new SizingConfig(
+ Integer.parseInt(configValue(props, minCountProp)),
+ Integer.parseInt(configValue(props, maxCountProp)),
+ Float.parseFloat(configValue(props,
HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP)),
+ Long.parseLong(configValue(props,
HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP)));
+
+ HealthCheckResult.Builder result = HealthCheckResult.builder(NAME)
+ .config("observed.index.layout", partitionedIndex ? "partitioned" :
"global")
+ .config("observed.index.layout.source",
+ layoutOnStorage.isPresent() ? "index definition on storage" :
"writer properties")
+ .config(minCountProp.key(), sizing.minFileGroupCount)
+ .config(maxCountProp.key(), sizing.maxFileGroupCount)
+
.config(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key(),
sizing.maxFileGroupSizeBytes)
+ .config(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key(),
sizing.growthFactor)
+ .config("effective.average.record.size.bytes",
RECORD_INDEX_AVERAGE_RECORD_SIZE);
+
+ // A writer configured for a layout the table does not have is worth
saying out loud: the index
+ // keeps the shape it was initialized with, so the writer's sizing keys
are not the ones
+ // governing it.
+ if (layoutOnStorage.isPresent()
+ &&
props.containsKey(HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key())
+ && ConfigUtils.getBooleanWithAltKeys(props,
HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP)
+ != layoutOnStorage.get()) {
+ result.finding(String.format(
+ "The index on storage is %s, but the writer sets %s=%s. The layout
is fixed at "
+ + "initialization, so the index keeps the shape it has and the
%s sizing keys are the "
+ + "ones governing it.",
+ partitionedIndex ? "partitioned" : "global",
+ HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
!layoutOnStorage.get(),
+ partitionedIndex ? "partitioned" : "global"));
+ }
+
+ if (sizing.maxFileGroupSizeBytes < RECORD_INDEX_AVERAGE_RECORD_SIZE) {
+ return result.status(HealthStatus.SKIPPED)
+ .summary(String.format(
+ "'%s' is %d bytes, smaller than the %d-byte average index record
the writer sizes with, so no "
+ + "file-group count can be estimated. The writer cannot size
an index with this value either.",
+
HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key(),
sizing.maxFileGroupSizeBytes,
+ RECORD_INDEX_AVERAGE_RECORD_SIZE))
+ .build();
+ }
+
+ RecordIndexFootprint footprint =
measureRecordIndex(context.getMetaClient());
+ long maxHealthyFileGroupBytes = (long) (MAX_FILE_GROUP_SIZE_SLACK_FACTOR *
sizing.maxFileGroupSizeBytes);
+ result.config("effective.max.healthy.file.group.bytes",
maxHealthyFileGroupBytes)
+ .config("observed.file.group.count", footprint.fileGroupCount)
+ .config("observed.base.file.count", footprint.baseFileCount)
+ .config("observed.log.file.count", footprint.logFileCount)
+ .config("observed.total.size.bytes", footprint.totalSizeBytes)
+ .config("observed.base.file.size.bytes", footprint.baseFileSizeBytes)
+ .config("effective.estimate.basis", footprint.estimateBasis())
+ .config("observed.largest.file.group.bytes",
footprint.largestFileGroupBytes)
+ .config("observed.largest.file.group.id",
footprint.largestFileGroupId);
+
+ Option<CountComparison> comparison = compareFileGroupCount(footprint,
partitionedIndex, sizing, result);
+ boolean undersized = comparison.isPresent() &&
comparison.get().isUndersized();
+ boolean oversizedFileGroup = footprint.largestFileGroupBytes >
maxHealthyFileGroupBytes;
+
+ if (!undersized && !oversizedFileGroup) {
+ HealthCheckResult.Builder healthy = result.status(HealthStatus.HEALTHY);
+ if (comparison.isPresent() && comparison.get().idealIsFloor) {
+ CountComparison floorBound = comparison.get();
+ // Report the floor rather than passing silently: the index is below
the configured
+ // minimum, and an operator who looks at the raw counts should find
out here why that is
+ // not a problem, instead of reading it as one the check missed.
+ healthy.summary(String.format(
+ "%s has %d file group(s), fewer than the configured minimum of %d,
but its estimated %d records are "
+ + "well inside what one file group holds, so the minimum is
the only thing calling for more",
+ floorBound.subject(), floorBound.observedFileGroupCount,
sizing.minFileGroupCount,
+ floorBound.estimatedRecordCount));
+ healthy.finding(String.format(
+ "The writer's sizing resolves to %d file group(s) for this index,
but that number comes from the "
+ + "configured minimum rather than from the table: at %d bytes
on storage the estimate is below one "
+ + "file group's worth of records. There is no shortfall to act
on, and re-bootstrapping to reach "
+ + "the minimum would buy nothing.",
+ floorBound.idealFileGroupCount, floorBound.observedSizeBytes));
+ return healthy.build();
+ }
+ return healthy
+ .summary(comparison.isPresent()
+ ? String.format("%s has %d file group(s) against an ideal of %d
for an estimated %d records; largest "
+ + "file group is %d bytes", comparison.get().subject(),
comparison.get().observedFileGroupCount,
+ comparison.get().idealFileGroupCount,
comparison.get().estimatedRecordCount, footprint.largestFileGroupBytes)
+ : String.format("Record index has %d file group(s); largest file
group is %d bytes, within the "
+ + "configured maximum. File-group count was not compared
against the writer's sizing.",
+ footprint.fileGroupCount, footprint.largestFileGroupBytes))
+ .build();
+ }
+
+ result.status(HealthStatus.UNHEALTHY);
+ if (undersized) {
+ CountComparison shortfall = comparison.get();
+ result.summary(String.format(
+ "%s has %d file group(s) but the writer's sizing calls for %d
(at least %d to be healthy)",
+ shortfall.subject(), shortfall.observedFileGroupCount,
shortfall.idealFileGroupCount,
+ shortfall.minHealthyFileGroupCount()))
+ .finding(String.format(
+ "%s was initialized with %d file group(s). Its %d bytes of %s
put it at an estimated %d records "
+ + "(an estimate: those bytes divided by the %d-byte average
record the writer sizes with), which "
+ + "with the configured growth factor and maximum file-group
size calls for %d. Today each file "
+ + "group averages %.2fx of the configured maximum file-group
size; once the configured growth "
+ + "factor is applied, each is projected to hold roughly
%.1fx more keys than the sizing intends, "
+ + "so index lookups read larger files and every index write
concentrates on the same few file groups.",
+ shortfall.subject(), shortfall.observedFileGroupCount,
shortfall.estimateBytes, shortfall.estimateBasis,
+ shortfall.estimatedRecordCount,
RECORD_INDEX_AVERAGE_RECORD_SIZE, shortfall.idealFileGroupCount,
+ shortfall.averageFileGroupBytes() / (double)
sizing.maxFileGroupSizeBytes,
+ (double) shortfall.idealFileGroupCount /
Math.max(shortfall.observedFileGroupCount, 1)));
+ }
+ if (oversizedFileGroup) {
+ if (!undersized) {
+ result.summary(String.format(
+ "Record index file group %s is %d bytes, past the configured
maximum of %d bytes",
+ footprint.largestFileGroupId, footprint.largestFileGroupBytes,
sizing.maxFileGroupSizeBytes));
+ }
+ result.finding(String.format(
+ "File group %s is %d bytes, past the configured maximum of %d bytes
even with %.1fx slack. Compacting "
+ + "it takes proportionally longer and every lookup routed to it
reads that much more. A file group "
+ + "that keeps growing between metadata compactions means the
index has too few file groups for "
+ + "the table.",
+ footprint.largestFileGroupId, footprint.largestFileGroupBytes,
sizing.maxFileGroupSizeBytes,
+ MAX_FILE_GROUP_SIZE_SLACK_FACTOR));
+ }
+ result.finding(rebootstrapRemedy(partitionedIndex));
+ return result.build();
+ }
+
+ /**
+ * Compares the file-group count against the writer's sizing estimate, at
the granularity the
+ * writer sizes at: the whole index for a global record index, each data
partition for a
+ * partitioned one (the worst data partition is returned). Echoes what it
compared, or why it
+ * did not, into the result. Empty when the writer's configured layout
disagrees with the layout
+ * on storage, since neither estimate would be comparable to the observed
count then.
+ */
+ static Option<CountComparison> compareFileGroupCount(RecordIndexFootprint
footprint, boolean partitionedIndex,
+ SizingConfig sizing,
HealthCheckResult.Builder result) {
+ boolean layoutKnown = footprint.fileGroupCount > 0;
+ if (layoutKnown && footprint.partitionedLayout != partitionedIndex) {
+ result.config("effective.count.comparison", String.format(
+ "skipped: writer configuration says a %s record index but the index
on storage is %s",
+ partitionedIndex ? "partitioned" : "global",
footprint.partitionedLayout ? "partitioned" : "global"));
+ return Option.empty();
+ }
+
+ CountComparison comparison;
+ if (!partitionedIndex) {
+ result.config("effective.count.comparison", "whole index (global record
index)");
+ comparison = CountComparison.of(Option.empty(), footprint, sizing);
+ } else {
+ result.config("effective.count.comparison", "per data partition
(partitioned record index); worst data partition reported")
+ .config("observed.data.partition.count",
footprint.byDataPartition.size());
+ // The worst data partition is the one the verdict should turn on, which
is not the one with the
+ // lowest ratio: a partition that is floor-bound has a low ratio only
because the configured
+ // minimum is high, and must not win the pick over a partition with a
real shortfall. So an
+ // undersized partition outranks a healthy one, a size-derived ideal
outranks a floor-derived
+ // one, and the ratio is the tiebreak within a class.
+ CountComparison worst = null;
+ int undersizedDataPartitions = 0;
+ for (Map.Entry<String, RecordIndexFootprint> entry :
footprint.byDataPartition.entrySet()) {
+ CountComparison candidate =
CountComparison.of(Option.of(entry.getKey()), entry.getValue(), sizing);
+ if (candidate.isUndersized()) {
+ undersizedDataPartitions++;
+ }
+ if (worst == null || candidate.isWorseThan(worst)) {
+ worst = candidate;
+ }
+ }
+ comparison = worst != null ? worst : CountComparison.of(Option.empty(),
footprint, sizing);
+ result.config("observed.undersized.data.partition.count",
undersizedDataPartitions);
+ if (comparison.dataPartition.isPresent()) {
+ result.config("observed.worst.data.partition",
comparison.dataPartition.get())
+ .config("observed.worst.data.partition.file.group.count",
comparison.observedFileGroupCount);
+ }
+ }
+
+ result.config("effective.ideal.file.group.count",
comparison.idealFileGroupCount)
+ .config("effective.ideal.source", comparison.idealIsFloor ?
"configured minimum (floor)" : "size estimate")
+ .config("effective.min.healthy.file.group.count",
comparison.minHealthyFileGroupCount())
+ .config("observed.estimated.record.count",
comparison.estimatedRecordCount);
+ return Option.of(comparison);
+ }
+
+ /**
+ * The one remedy for a mis-sized record index. The file-group count is
fixed at initialization,
+ * so there is no in-place resize; the operator has to drop the index and
build it again.
+ *
+ * <p>The rebuild path differs by layout. {@code HoodieIndexer} cannot
currently rebuild a
+ * partitioned record index -- see HUDI issue #20343 -- so for that layout
the remedy says what
+ * is broken rather than naming a command that silently produces a global
index instead.
+ */
+ private static String rebootstrapRemedy(boolean partitionedIndex) {
+ String preamble = "The file-group count of the record index is fixed when
the index is initialized and cannot "
+ + "be changed in place. Resizing it means dropping the index and
building it again, which is disruptive: "
+ + "the record index is unavailable until the rebuild completes, and
the drop cannot be undone in place. "
+ + "Plan it as a maintenance window.";
+ if (partitionedIndex) {
+ return preamble + " Two things make a rebuild the wrong move here.
HoodieIndexer cannot currently rebuild a "
+ + "partitioned record index: it maps the on-storage 'record_index'
partition to the global enable flag "
+ + "unconditionally, so a rebuild produces a global index with
default sizing, and supplying the "
+ + "partitioned key in its own --props does not help because that
mapping overwrites it "
+ + "(https://github.com/apache/hudi/issues/20343). And a rebuild
would not hold anyway: data partitions "
+ + "added afterwards are always initialized at "
+ +
HoodieMetadataConfig.RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key() + "
file groups, never from the "
+ + "size estimate, so tomorrow's partition starts undersized again.
Raising that minimum is the lever that "
+ + "persists; it applies to partitions created from then on, not to
the ones already initialized.";
+ }
+ return preamble + " Drop it with 'metadata delete-record-index' in
hudi-cli (or HoodieIndexer --mode dropindex "
+ + "--index-types RECORD_INDEX; note the hudi-cli command takes a
backup by default), correct the sizing "
+ + "configuration, then rebuild it with HoodieIndexer --mode
scheduleAndExecute --index-types RECORD_INDEX.";
+ }
+
+ /**
+ * The index layout as recorded on storage, when the table has an index
definition for the record
+ * index. {@code HoodieRecordIndex.IS_PARTITIONED_OPTION} is written at
initialization, so it
+ * describes the index that exists rather than the one a writer is
configured for.
+ *
+ * @return empty when no record-index definition is present, in which case
the caller falls back
+ * to the writer property.
+ */
+ private static Option<Boolean>
partitionedIndexOnStorage(HoodieTableMetaClient metaClient) {
+ return metaClient.getIndexMetadata()
+ .map(metadata ->
metadata.getIndexDefinitions().get(PARTITION_NAME_RECORD_INDEX))
+ .map(HoodieRecordIndex::isPartitioned);
+ }
+
+ /**
+ * Like {@link HealthCheckContext#firstMissingConfig(String...)}, but aware
of each property's
+ * alternative keys, since the record-index sizing keys were renamed and a
writer may still supply
+ * the old spelling.
+ */
+ private static Option<String> firstMissingConfig(HealthCheckContext context,
ConfigProperty<?>... properties) {
+ if (context.isApplyAllDefaults()) {
+ return Option.empty();
+ }
+ for (ConfigProperty<?> property : properties) {
+ if (!ConfigUtils.containsConfigProperty(context.getProps(), property)) {
+ return Option.of(property.key());
+ }
+ }
+ return Option.empty();
+ }
+
+ private static String configValue(TypedProperties props, ConfigProperty<?>
property) {
+ return ConfigUtils.getStringWithAltKeys(props, property,
String.valueOf(property.defaultValue()));
+ }
+
+ /**
+ * Sizes the record index from the latest merged file slice of each of its
file groups in the
+ * metadata table, and per data partition as well when the file-group names
show a partitioned
+ * layout. The metadata reader is engine-agnostic and closed once the slices
are read.
+ */
+ private static RecordIndexFootprint measureRecordIndex(HoodieTableMetaClient
metaClient) throws Exception {
+ HoodieLocalEngineContext engineContext = new
HoodieLocalEngineContext(metaClient.getStorageConf());
+ HoodieMetadataConfig metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true).build();
+ try (HoodieBackedTableMetadata metadata = new HoodieBackedTableMetadata(
+ engineContext, metaClient.getStorage(), metadataConfig,
metaClient.getBasePath().toString())) {
+ List<FileSlice> fileSlices =
metadata.getFilegroupsForPartition(MetadataPartitionType.RECORD_INDEX);
+ boolean partitionedLayout = !fileSlices.isEmpty()
+ &&
HoodieTableMetadataUtil.verifyRLIFile(fileSlices.get(0).getFileId(), true);
+
+ Map<String, RecordIndexFootprint> byDataPartition = new TreeMap<>();
+ if (partitionedLayout) {
+
metadata.getBucketizedFileGroupsForPartitionedRLI(MetadataPartitionType.RECORD_INDEX)
+ .forEach((dataPartition, slices) ->
byDataPartition.put(dataPartition, RecordIndexFootprint.of(slices)));
+ }
+ return RecordIndexFootprint.of(fileSlices).withLayout(partitionedLayout,
byDataPartition);
+ }
+ }
+
+ /** The writer's sizing configuration, as supplied or defaulted. */
+ static final class SizingConfig {
+ final int minFileGroupCount;
+ final int maxFileGroupCount;
+ final float growthFactor;
+ final long maxFileGroupSizeBytes;
+
+ SizingConfig(int minFileGroupCount, int maxFileGroupCount, float
growthFactor, long maxFileGroupSizeBytes) {
+ this.minFileGroupCount = minFileGroupCount;
+ this.maxFileGroupCount = maxFileGroupCount;
+ this.growthFactor = growthFactor;
+ this.maxFileGroupSizeBytes = maxFileGroupSizeBytes;
+ }
+ }
+
+ /** One observed-versus-ideal file-group comparison, for the whole index or
one data partition. */
+ static final class CountComparison {
+ final Option<String> dataPartition;
+ final int observedFileGroupCount;
+ final long observedSizeBytes;
+ /** The bytes the record estimate was derived from, and what they were:
see {@link RecordIndexFootprint#estimateBytes()}. */
+ final long estimateBytes;
+ final String estimateBasis;
+ final long estimatedRecordCount;
+ final int idealFileGroupCount;
+ /**
+ * Whether {@link #idealFileGroupCount} came from the configured floor
rather than from the
+ * table's own size. When it did, the ideal carries no information about
how full the index
+ * actually is, so a shortfall against it says nothing about index health.
+ */
+ final boolean idealIsFloor;
+
+ private CountComparison(Option<String> dataPartition, int
observedFileGroupCount, long observedSizeBytes,
+ long estimateBytes, String estimateBasis, long
estimatedRecordCount,
+ int idealFileGroupCount, boolean idealIsFloor) {
+ this.dataPartition = dataPartition;
+ this.observedFileGroupCount = observedFileGroupCount;
+ this.observedSizeBytes = observedSizeBytes;
+ this.estimateBytes = estimateBytes;
+ this.estimateBasis = estimateBasis;
+ this.estimatedRecordCount = estimatedRecordCount;
+ this.idealFileGroupCount = idealFileGroupCount;
+ this.idealIsFloor = idealIsFloor;
+ }
+
+ static CountComparison of(Option<String> dataPartition,
RecordIndexFootprint footprint, SizingConfig sizing) {
+ long estimateBytes = footprint.estimateBytes();
+ long estimatedRecordCount = estimateBytes /
RECORD_INDEX_AVERAGE_RECORD_SIZE;
+ int idealFileGroupCount = HoodieTableMetadataUtil.estimateFileGroupCount(
+ MetadataPartitionType.RECORD_INDEX, () -> estimatedRecordCount,
RECORD_INDEX_AVERAGE_RECORD_SIZE,
+ sizing.minFileGroupCount, sizing.maxFileGroupCount,
sizing.growthFactor, sizing.maxFileGroupSizeBytes);
+ return new CountComparison(dataPartition, footprint.fileGroupCount,
footprint.totalSizeBytes, estimateBytes,
+ footprint.estimateBasis(), estimatedRecordCount, idealFileGroupCount,
+ isFloorBound(estimatedRecordCount, sizing));
+ }
+
+ /**
+ * Mirrors the two branches in {@code
HoodieTableMetadataUtil.estimateFileGroupCount} that
+ * return the minimum without regard to table size: a fixed count, where
min and max are equal,
+ * and an estimate at or below the minimum. A 3 MB index against a 1 GiB
maximum file-group
+ * size estimates zero file groups and is clamped up to the floor, which
is how a tiny index
+ * ends up being told it needs ten.
+ */
+ private static boolean isFloorBound(long estimatedRecordCount,
SizingConfig sizing) {
+ if (sizing.minFileGroupCount == sizing.maxFileGroupCount &&
sizing.minFileGroupCount != 0) {
+ return true;
+ }
+ long expectedNumRecords = (long) Math.ceil(estimatedRecordCount *
(double) sizing.growthFactor);
+ long maxRecordsPerFileGroup = sizing.maxFileGroupSizeBytes /
Math.max(RECORD_INDEX_AVERAGE_RECORD_SIZE, 1L);
+ long estimatedFileGroupCount = expectedNumRecords /
Math.max(maxRecordsPerFileGroup, 1L);
+ return estimatedFileGroupCount <= sizing.minFileGroupCount;
+ }
+
+ int minHealthyFileGroupCount() {
+ return (int) Math.ceil(MIN_ACCEPTABLE_FILE_GROUP_RATIO *
idealFileGroupCount);
+ }
+
+ boolean isUndersized() {
+ // A shortfall against a floor-derived ideal is not evidence of
anything: the index is below
+ // the configured minimum, but the estimate says it is nowhere near a
single file group's
+ // capacity, so there is no sizing problem to re-bootstrap away.
+ return !idealIsFloor && observedFileGroupCount <
minHealthyFileGroupCount();
+ }
+
+ double ratio() {
+ return (double) observedFileGroupCount / Math.max(idealFileGroupCount,
1);
+ }
+
+ long averageFileGroupBytes() {
+ return observedSizeBytes / Math.max(observedFileGroupCount, 1);
+ }
+
+ /**
+ * Orders comparisons by what the verdict depends on. An undersized
comparison is worse than one
+ * that is not; between two that agree on that, a size-derived ideal is
worse than a floor-derived
+ * one, because only the former says anything about the index; and within
a class the lower
+ * observed-to-ideal ratio is worse. Ratio alone would let a floor-bound
partition -- low ratio,
+ * no shortfall -- win the pick and hide a partition with a real one.
+ */
+ boolean isWorseThan(CountComparison other) {
+ if (isUndersized() != other.isUndersized()) {
+ return isUndersized();
+ }
+ if (idealIsFloor != other.idealIsFloor) {
+ return !idealIsFloor;
+ }
+ return ratio() < other.ratio();
+ }
+
+ String subject() {
+ return dataPartition.map(p -> String.format("Data partition '%s' of the
record index", p))
+ .orElse("Record index");
+ }
+ }
+
+ /** What the record index occupies on storage, taken from its latest file
slices. */
+ static final class RecordIndexFootprint {
+ final int fileGroupCount;
+ final int baseFileCount;
+ final int logFileCount;
+ final long totalSizeBytes;
+ final long baseFileSizeBytes;
+ final long largestFileGroupBytes;
+ final String largestFileGroupId;
+ final boolean partitionedLayout;
+ final Map<String, RecordIndexFootprint> byDataPartition;
+
+ private RecordIndexFootprint(int fileGroupCount, int baseFileCount, int
logFileCount, long totalSizeBytes,
+ long baseFileSizeBytes, long
largestFileGroupBytes, String largestFileGroupId,
+ boolean partitionedLayout, Map<String,
RecordIndexFootprint> byDataPartition) {
+ this.fileGroupCount = fileGroupCount;
+ this.baseFileCount = baseFileCount;
+ this.logFileCount = logFileCount;
+ this.totalSizeBytes = totalSizeBytes;
+ this.baseFileSizeBytes = baseFileSizeBytes;
+ this.largestFileGroupBytes = largestFileGroupBytes;
+ this.largestFileGroupId = largestFileGroupId;
+ this.partitionedLayout = partitionedLayout;
+ this.byDataPartition = byDataPartition;
+ }
+
+ static RecordIndexFootprint of(List<FileSlice> fileSlices) {
+ int baseFiles = 0;
+ int logFiles = 0;
+ long totalBytes = 0;
+ long baseBytes = 0;
+ long largestBytes = 0;
+ String largestId = "";
+ for (FileSlice slice : fileSlices) {
+ baseFiles += slice.getBaseFile().isPresent() ? 1 : 0;
+ baseBytes +=
slice.getBaseFile().map(HoodieBaseFile::getFileSize).orElse(0L);
+ logFiles += slice.getLogFileCnt();
+ long sliceBytes = slice.getTotalFileSize();
+ totalBytes += sliceBytes;
+ if (sliceBytes > largestBytes) {
+ largestBytes = sliceBytes;
+ largestId = slice.getFileId();
+ }
+ }
+ return new RecordIndexFootprint(fileSlices.size(), baseFiles, logFiles,
totalBytes, baseBytes, largestBytes,
+ largestId, false, new TreeMap<>());
+ }
+
+ /** A footprint with the given measurements and no file slices behind it,
for tests. */
+ static RecordIndexFootprint synthetic(int fileGroupCount, int
baseFileCount, long baseFileSizeBytes,
+ long totalSizeBytes) {
+ return new RecordIndexFootprint(fileGroupCount, baseFileCount,
fileGroupCount - baseFileCount, totalSizeBytes,
+ baseFileSizeBytes, totalSizeBytes / Math.max(fileGroupCount, 1),
"synthetic", false, new TreeMap<>());
+ }
+
+ RecordIndexFootprint withLayout(boolean partitioned, Map<String,
RecordIndexFootprint> perDataPartition) {
+ return new RecordIndexFootprint(fileGroupCount, baseFileCount,
logFileCount, totalSizeBytes, baseFileSizeBytes,
+ largestFileGroupBytes, largestFileGroupId, partitioned,
perDataPartition);
+ }
+
+ /**
+ * The bytes the record estimate is derived from. Log files in the record
index carry updates
+ * and deletes for keys the base files already hold, so counting them
inflates the estimate
+ * without any new records behind it; the base files, written by metadata
compaction, are the
+ * deduplicated set. Before the first compaction there are no base files,
and the log files are
+ * the only measurement there is.
+ */
+ long estimateBytes() {
+ return baseFileCount > 0 ? baseFileSizeBytes : totalSizeBytes;
+ }
+
+ String estimateBasis() {
+ return baseFileCount > 0 ? "base files" : "log files (the index has no
base file yet)";
+ }
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
index 9bd07ef32fe8..6b9a9ec853f4 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
@@ -215,7 +215,7 @@ public class TestHoodieTableHealthChecker {
JsonNode root = MAPPER.readTree(result.output);
assertEquals("UNHEALTHY", root.get("overallStatus").asText());
assertEquals(HoodieTableType.MERGE_ON_READ.name(),
root.get("tableType").asText());
- assertEquals(6, root.get("checks").size());
+ assertEquals(7, root.get("checks").size());
JsonNode compaction = root.get("checks").get(0);
assertEquals("compaction", compaction.get("name").asText());
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestRecordIndexSizingCheck.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestRecordIndexSizingCheck.java
new file mode 100644
index 000000000000..6352a02c2542
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestRecordIndexSizingCheck.java
@@ -0,0 +1,298 @@
+/*
+ * 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.health;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.metadata.MetadataPartitionType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.TreeMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests the paths of {@link RecordIndexSizingCheck} that need no metadata
table on disk: the check
+ * declining to run, and the count comparison on synthetic footprints, where
the healthy/undersized
+ * boundary and the floor rule can be placed exactly. Verdicts on a real
record index are covered by
+ * {@link TestRecordIndexSizingCheckWithSpark}, which has to write one.
+ */
+public class TestRecordIndexSizingCheck {
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+ private final RecordIndexSizingCheck check = new RecordIndexSizingCheck();
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ private HoodieTableMetaClient tableWithRecordIndexListed() throws Exception {
+ // Only the table configuration says the partition exists; there is no
metadata table behind it.
+ // Enough for the check's fail-fast paths, which must not touch the
metadata table.
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath,
HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable.of(metaClient).addCommit("001");
+ metaClient.getTableConfig().setMetadataPartitionState(
+ metaClient, MetadataPartitionType.RECORD_INDEX.getPartitionPath(),
true);
+ return HoodieTableMetaClient.reload(metaClient);
+ }
+
+ private TypedProperties globalSizingProps() {
+ TypedProperties props = new TypedProperties();
+
props.setProperty(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key(),
"2");
+
props.setProperty(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(),
"2");
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key(),
"1073741824");
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key(),
"2.0");
+ return props;
+ }
+
+ @Test
+ public void tableWithoutRecordIndexIsSkippedAndSaysSo() throws Exception {
+ // given a table whose metadata table has no record index partition
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath,
HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable.of(metaClient).addCommit("001");
+
+ // when the check runs with every sizing property supplied
+ HealthCheckResult result = check.check(new HealthCheckContext(metaClient,
globalSizingProps(), false));
+
+ // then it does not apply, and the reason names the partition it looked for
+ assertFalse(check.appliesTo(new HealthCheckContext(metaClient,
globalSizingProps(), false)));
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+
assertTrue(result.getSummary().contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath()),
+ "expected the skip reason to name the record index partition, got: " +
result.getSummary());
+ }
+
+ @Test
+ public void missingSizingConfigSkipsNamingTheFirstMissingProperty() throws
Exception {
+ // given a table with a record index and no writer configuration supplied
+ HoodieTableMetaClient metaClient = tableWithRecordIndexListed();
+
+ // when the check runs
+ HealthCheckResult result = check.check(new HealthCheckContext(metaClient,
new TypedProperties(), false));
+
+ // then it declines to judge and names the global minimum file-group count
first
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains(
+
HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key()),
result.getSummary());
+ }
+
+ @Test
+ public void eachSizingPropertyIsRequiredInTurn() throws Exception {
+ // given a record index and a properties set missing only the growth factor
+ HoodieTableMetaClient metaClient = tableWithRecordIndexListed();
+ TypedProperties props = globalSizingProps();
+ props.remove(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key());
+
+ // when the check runs
+ HealthCheckResult result = check.check(new HealthCheckContext(metaClient,
props, false));
+
+ // then that is the property it asks for
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+
assertTrue(result.getSummary().contains(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key()),
+ result.getSummary());
+ }
+
+ @Test
+ public void partitionedRecordIndexRequiresThePartitionedSizingKeys() throws
Exception {
+ // given a writer that enables the partitioned record index but only
supplies the global sizing keys
+ HoodieTableMetaClient metaClient = tableWithRecordIndexListed();
+ TypedProperties props = globalSizingProps();
+
props.setProperty(HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
"true");
+
+ // when the check runs
+ HealthCheckResult result = check.check(new HealthCheckContext(metaClient,
props, false));
+
+ // then it asks for the partitioned key, since that is what the writer
sizes with
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains(
+
HoodieMetadataConfig.RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key()),
result.getSummary());
+ }
+
+ @Test
+ public void
maxFileGroupSizeBelowOneRecordIsRejectedRatherThanDividedByZero() throws
Exception {
+ // given a maximum file-group size smaller than a single index record
+ HoodieTableMetaClient metaClient = tableWithRecordIndexListed();
+ TypedProperties props = globalSizingProps();
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key(),
"10");
+
+ // when the check runs
+ HealthCheckResult result = check.check(new HealthCheckContext(metaClient,
props, false));
+
+ // then it skips with the offending value echoed, before ever opening the
metadata table
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+
assertTrue(result.getSummary().contains(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key()),
+ result.getSummary());
+ assertEquals("10", result.getEffectiveConfigs()
+
.get(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key()));
+ }
+
+ // ---- count comparison on synthetic footprints
-------------------------------------------------
+
+ private static final long ONE_GIB = 1024L * 1024 * 1024;
+ /** Records that fit one 1 GiB file group at the writer's 48-byte average
record. */
+ private static final long RECORDS_PER_FILE_GROUP = ONE_GIB / 48;
+
+ private static RecordIndexSizingCheck.SizingConfig sizing(int minFileGroups,
int maxFileGroups) {
+ return new RecordIndexSizingCheck.SizingConfig(minFileGroups,
maxFileGroups, 2.0f, ONE_GIB);
+ }
+
+ /**
+ * Base-file bytes that make the writer's estimate land on {@code
idealFileGroups}: the estimate
+ * multiplies the record count by the 2.0 growth factor and divides by the
records per file group,
+ * so half that many records' worth of bytes per ideal file group. Aimed at
the middle of the band
+ * (ideal + 0.5) rather than its lower edge, because the writer does that
multiplication in
+ * {@code float} and a count above 2^24 records rounds by a few either way.
+ */
+ private static long bytesForIdeal(int idealFileGroups) {
+ return (long) ((idealFileGroups + 0.5) * RECORDS_PER_FILE_GROUP / 2) * 48;
+ }
+
+ private static RecordIndexSizingCheck.RecordIndexFootprint footprint(int
fileGroups, long baseFileBytes) {
+ return RecordIndexSizingCheck.RecordIndexFootprint.synthetic(fileGroups,
fileGroups, baseFileBytes, baseFileBytes);
+ }
+
+ private static RecordIndexSizingCheck.CountComparison compare(int
fileGroups, long baseFileBytes,
+
RecordIndexSizingCheck.SizingConfig sizing) {
+ return RecordIndexSizingCheck.CountComparison.of(Option.empty(),
footprint(fileGroups, baseFileBytes), sizing);
+ }
+
+ @Test
+ public void fileGroupsAtExactlyHalfTheIdealAreHealthy() {
+ // given an index whose bytes call for four file groups and which has two
+ RecordIndexSizingCheck.CountComparison comparison = compare(2,
bytesForIdeal(4), sizing(1, 1000));
+
+ // then the ideal is size-derived, the ratio sits exactly on the 0.5
boundary, and it is not undersized
+ assertEquals(4, comparison.idealFileGroupCount);
+ assertFalse(comparison.idealIsFloor);
+ assertEquals(2, comparison.minHealthyFileGroupCount());
+ assertEquals(0.5, comparison.ratio(), 1e-9);
+ assertFalse(comparison.isUndersized());
+ }
+
+ @Test
+ public void fileGroupsJustUnderHalfTheIdealAreUndersized() {
+ // given the same sizing with one file group fewer, and a five-ideal index
with two
+ RecordIndexSizingCheck.CountComparison oneOfFour = compare(1,
bytesForIdeal(4), sizing(1, 1000));
+ RecordIndexSizingCheck.CountComparison twoOfFive = compare(2,
bytesForIdeal(5), sizing(1, 1000));
+
+ // then both fall below ceil(0.5 * ideal) and are undersized
+ assertEquals(4, oneOfFour.idealFileGroupCount);
+ assertTrue(oneOfFour.isUndersized());
+ assertEquals(5, twoOfFive.idealFileGroupCount);
+ assertEquals(3, twoOfFive.minHealthyFileGroupCount());
+ assertTrue(twoOfFive.isUndersized());
+ }
+
+ @Test
+ public void floorBoundIdealIsNeverUndersized() {
+ // given a fixed count (min == max) and a minimum far above what a tiny
index needs
+ RecordIndexSizingCheck.CountComparison fixedCount = compare(1,
bytesForIdeal(4), sizing(4, 4));
+ RecordIndexSizingCheck.CountComparison tinyIndex = compare(1, 1024 * 1024,
sizing(10, 1000));
+
+ // then both ideals come from the floor and neither comparison reports a
shortfall, however low the ratio
+ assertTrue(fixedCount.idealIsFloor);
+ assertEquals(0.25, fixedCount.ratio(), 1e-9);
+ assertFalse(fixedCount.isUndersized());
+ assertTrue(tinyIndex.idealIsFloor);
+ assertEquals(10, tinyIndex.idealFileGroupCount);
+ assertEquals(0.1, tinyIndex.ratio(), 1e-9);
+ assertFalse(tinyIndex.isUndersized());
+ }
+
+ @Test
+ public void worstDataPartitionIsTheOneWithARealShortfallNotTheLowestRatio() {
+ // given a partitioned index after the minimum was raised from 1 to 10: an
old one-file-group
+ // partition is floor-bound at ratio 0.1, and a large partition has five
file groups against a
+ // size-derived ideal of twenty, ratio 0.25
+ RecordIndexSizingCheck.SizingConfig sizing = sizing(10, 1000);
+ Map<String, RecordIndexSizingCheck.RecordIndexFootprint> byDataPartition =
new TreeMap<>();
+ byDataPartition.put("2024/01/01", footprint(1, 1024 * 1024));
+ byDataPartition.put("2024/01/02", footprint(5, bytesForIdeal(20)));
+ RecordIndexSizingCheck.RecordIndexFootprint index = footprint(6,
bytesForIdeal(20) + 1024 * 1024)
+ .withLayout(true, byDataPartition);
+ HealthCheckResult.Builder result =
HealthCheckResult.builder(RecordIndexSizingCheck.NAME);
+
+ // when the worst data partition is picked
+ RecordIndexSizingCheck.CountComparison worst =
+ RecordIndexSizingCheck.compareFileGroupCount(index, true, sizing,
result).get();
+
+ // then the real shortfall wins the pick over the lower ratio, and the
verdict is undersized
+ assertEquals("2024/01/02", worst.dataPartition.get());
+ assertEquals(20, worst.idealFileGroupCount);
+ assertFalse(worst.idealIsFloor);
+ assertTrue(worst.isUndersized());
+ Map<String, String> configs = result.build().getEffectiveConfigs();
+ assertEquals("2024/01/02", configs.get("observed.worst.data.partition"));
+ assertEquals("1", configs.get("observed.undersized.data.partition.count"));
+ assertEquals("size estimate", configs.get("effective.ideal.source"));
+ }
+
+ @Test
+ public void amongHealthyDataPartitionsTheLowestSizeDerivedRatioIsReported() {
+ // given no shortfall anywhere: one floor-bound partition at ratio 0.1 and
two size-derived
+ // partitions at ratios 1.0 and 0.75
+ RecordIndexSizingCheck.SizingConfig sizing = sizing(10, 1000);
+ Map<String, RecordIndexSizingCheck.RecordIndexFootprint> byDataPartition =
new TreeMap<>();
+ byDataPartition.put("a", footprint(1, 1024 * 1024));
+ byDataPartition.put("b", footprint(20, bytesForIdeal(20)));
+ byDataPartition.put("c", footprint(15, bytesForIdeal(20)));
+ RecordIndexSizingCheck.RecordIndexFootprint index = footprint(36, 2 *
bytesForIdeal(20)).withLayout(true, byDataPartition);
+ HealthCheckResult.Builder result =
HealthCheckResult.builder(RecordIndexSizingCheck.NAME);
+
+ // when the worst data partition is picked
+ RecordIndexSizingCheck.CountComparison worst =
+ RecordIndexSizingCheck.compareFileGroupCount(index, true, sizing,
result).get();
+
+ // then the floor-bound partition does not win on ratio; the size-derived
0.75 does, and nothing is undersized
+ assertEquals("c", worst.dataPartition.get());
+ assertFalse(worst.isUndersized());
+ assertEquals("0",
result.build().getEffectiveConfigs().get("observed.undersized.data.partition.count"));
+ }
+
+ @Test
+ public void
recordEstimateUsesBaseFileBytesAndFallsBackToLogsBeforeTheFirstCompaction() {
+ // given a compacted index whose log files triple its footprint, and an
uncompacted one with logs only
+ RecordIndexSizingCheck.RecordIndexFootprint compacted =
+ RecordIndexSizingCheck.RecordIndexFootprint.synthetic(2, 2, ONE_GIB, 3
* ONE_GIB);
+ RecordIndexSizingCheck.RecordIndexFootprint logsOnly =
+ RecordIndexSizingCheck.RecordIndexFootprint.synthetic(2, 0, 0,
ONE_GIB);
+
+ // then the estimate ignores the log bytes when base files exist, and uses
them when nothing else is there
+ assertEquals(ONE_GIB, compacted.estimateBytes());
+ assertEquals("base files", compacted.estimateBasis());
+ assertEquals(ONE_GIB / 48,
RecordIndexSizingCheck.CountComparison.of(Option.empty(), compacted, sizing(1,
1000)).estimatedRecordCount);
+ assertEquals(ONE_GIB, logsOnly.estimateBytes());
+ assertEquals("log files (the index has no base file yet)",
logsOnly.estimateBasis());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestRecordIndexSizingCheckWithSpark.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestRecordIndexSizingCheckWithSpark.java
new file mode 100644
index 000000000000..22b13ffdc3c1
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestRecordIndexSizingCheckWithSpark.java
@@ -0,0 +1,428 @@
+/*
+ * 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.health;
+
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link RecordIndexSizingCheck} verdicts against a record index
actually written through
+ * the Spark write client. The global index is initialized with a fixed two
file groups and the
+ * partitioned one with a fixed one file group per data partition, so the
check's verdict is driven
+ * entirely by the sizing configuration handed to it.
+ */
+public class TestRecordIndexSizingCheckWithSpark extends
SparkClientFunctionalTestHarness {
+
+ private static final HoodieTestDataGenerator DATA_GENERATOR = new
HoodieTestDataGenerator(0L);
+ private static final int GLOBAL_FILE_GROUPS = 2;
+ private static final int FILE_GROUPS_PER_DATA_PARTITION = 1;
+ private static final int DATA_PARTITIONS =
HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS.length;
+ private static final String ONE_GIB = String.valueOf(1024L * 1024 * 1024);
+ /**
+ * A maximum file-group size small enough that the fixture index estimates
several file groups
+ * rather than clamping to the configured minimum. With ONE_GIB the estimate
is zero for any
+ * fixture-sized index, so the ideal comes from the floor and says nothing
about the table --
+ * which is the bug these tests exist to catch, not a configuration to
assert against.
+ */
+ private static final String SIXTEEN_KIB = String.valueOf(16 * 1024);
+ /** Smaller still, so the estimate clears the 0.5 healthy ratio rather than
sitting on it. */
+ private static final String FOUR_KIB = String.valueOf(4 * 1024);
+ /**
+ * Small enough that one data partition's base-file bytes alone call for
several file groups. The
+ * estimate is taken from base files, not base plus logs, so the
per-partition fixture -- a few
+ * dozen records each -- needs a limit this low to show a real shortfall
rather than the floor.
+ */
+ private static final String ONE_KIB = String.valueOf(1024);
+
+ private final RecordIndexSizingCheck check = new RecordIndexSizingCheck();
+ private HoodieTableMetaClient metaClient;
+
+ @BeforeEach
+ public void initTable() throws Exception {
+ metaClient = getHoodieMetaClient(HoodieTableType.COPY_ON_WRITE);
+ }
+
+ @AfterAll
+ public static void closeDataGenerator() {
+ DATA_GENERATOR.close();
+ }
+
+ private void writeGlobalRecordIndex() throws Exception {
+ HoodieWriteConfig writeConfig = writeConfigBuilder()
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withEnableGlobalRecordLevelIndex(true)
+ .withRecordIndexFileGroupCount(GLOBAL_FILE_GROUPS,
GLOBAL_FILE_GROUPS)
+ .build())
+ .build();
+ insertBatches(writeConfig, 1);
+ }
+
+ private void writePartitionedRecordIndex() throws Exception {
+ // A partitioned record index is sized from the data partitions that
exist, so its
+ // initialization is deferred past the first commit and two batches are
written.
+ HoodieWriteConfig writeConfig = writeConfigBuilder()
+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.RECORD_LEVEL_INDEX).build())
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withEnableRecordLevelIndex(true)
+
.withPartitionedRecordIndexFileGroupCount(FILE_GROUPS_PER_DATA_PARTITION,
FILE_GROUPS_PER_DATA_PARTITION)
+ .withDeferRliInitializationForFreshTable(true)
+ .build())
+ .build();
+ insertBatches(writeConfig, 2);
+ }
+
+ private HoodieWriteConfig.Builder writeConfigBuilder() {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath())
+ .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA)
+ .withParallelism(2, 2)
+ .forTable("record_index_sizing");
+ }
+
+ private void insertBatches(HoodieWriteConfig writeConfig, int batches)
throws Exception {
+ try (SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context(),
writeConfig)) {
+ for (int i = 0; i < batches; i++) {
+ String instant = writeClient.startCommit();
+ List<HoodieRecord> records = DATA_GENERATOR.generateInserts(instant,
100);
+ List<WriteStatus> statuses =
writeClient.upsert(jsc().parallelize(records, 1), instant).collect();
+ assertNoWriteErrors(statuses);
+ writeClient.commit(instant, jsc().parallelize(statuses));
+ }
+ }
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
assertTrue(metaClient.getTableConfig().isMetadataPartitionAvailable(MetadataPartitionType.RECORD_INDEX),
+ "fixture should have initialized a record index");
+ }
+
+ private static TypedProperties globalSizingProps(int minFileGroups, int
maxFileGroups, String maxFileGroupSizeBytes) {
+ TypedProperties props = new TypedProperties();
+
props.setProperty(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key(),
+ String.valueOf(minFileGroups));
+
props.setProperty(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(),
+ String.valueOf(maxFileGroups));
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key(),
maxFileGroupSizeBytes);
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key(),
"2.0");
+ return props;
+ }
+
+ private static TypedProperties partitionedSizingProps(int minFileGroups, int
maxFileGroups, String maxFileGroupSizeBytes) {
+ TypedProperties props = new TypedProperties();
+
props.setProperty(HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
"true");
+
props.setProperty(HoodieMetadataConfig.RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key(),
+ String.valueOf(minFileGroups));
+
props.setProperty(HoodieMetadataConfig.RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(),
+ String.valueOf(maxFileGroups));
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key(),
maxFileGroupSizeBytes);
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key(),
"2.0");
+ return props;
+ }
+
+ private HealthCheckResult runCheck(TypedProperties props, boolean
applyAllDefaults) throws Exception {
+ return check.check(new HealthCheckContext(metaClient, props,
applyAllDefaults));
+ }
+
+ private static void assertCountComparisonAbsent(Map<String, String> configs)
{
+ assertFalse(configs.containsKey("effective.ideal.file.group.count"),
configs.toString());
+ assertFalse(configs.containsKey("effective.min.healthy.file.group.count"),
configs.toString());
+ assertFalse(configs.containsKey("observed.estimated.record.count"),
configs.toString());
+
assertTrue(configs.get("effective.count.comparison").startsWith("skipped:"),
configs.toString());
+ }
+
+ @Test
+ public void recordIndexMatchingItsConfiguredSizingIsHealthy() throws
Exception {
+ // given a global index sized so the estimate derives from the index's own
bytes rather than
+ // clamping to the floor, and lands at or below what the index already has
+ writeGlobalRecordIndex();
+ TypedProperties props = globalSizingProps(1, 64, SIXTEEN_KIB);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then the ideal is size-derived and the index is not called undersized
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(),
result.getSummary());
+ assertEquals("2",
result.getEffectiveConfigs().get("observed.file.group.count"));
+ assertEquals("whole index (global record index)",
result.getEffectiveConfigs().get("effective.count.comparison"));
+ int ideal =
Integer.parseInt(result.getEffectiveConfigs().get("effective.ideal.file.group.count"));
+ assertTrue(ideal > 1, "fixture should exercise the estimate, not the
floor; ideal was " + ideal);
+ }
+
+ @Test
+ public void
fewerFileGroupsThanTheWriterNowSizesForIsUnhealthyAndSaysToRebootstrap() throws
Exception {
+ // given a maximum file-group size small enough that the index's own bytes
call for more file
+ // groups than it was built with -- a real shortfall, not one manufactured
by the floor
+ writeGlobalRecordIndex();
+ TypedProperties props = globalSizingProps(1, 64, FOUR_KIB);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then it reports the index undersized, with the counts and the only
remedy spelled out
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus(),
result.getSummary());
+ int ideal =
Integer.parseInt(result.getEffectiveConfigs().get("effective.ideal.file.group.count"));
+ assertTrue(ideal > 1, "the ideal must come from the estimate, not the
floor; was " + ideal);
+ assertEquals(String.valueOf((int) Math.ceil(0.5 * ideal)),
+
result.getEffectiveConfigs().get("effective.min.healthy.file.group.count"));
+ assertEquals("2",
result.getEffectiveConfigs().get("observed.file.group.count"));
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("Record
index was initialized with 2 file group(s)")),
+ "expected a finding stating the observed count, got: " +
result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains("metadata delete-record-index")),
+ "expected the hudi-cli drop command, got: " + result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("--mode
dropindex --index-types RECORD_INDEX")),
+ "expected the HoodieIndexer drop command, got: " +
result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("--mode
scheduleAndExecute --index-types RECORD_INDEX")),
+ "expected the HoodieIndexer rebuild command, got: " +
result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("cannot
be changed in place")),
+ "expected the finding to say the count is fixed at initialization,
got: " + result.getFindings());
+ }
+
+ @Test
+ public void
fixedFileGroupCountAboveWhatExistsIsReportedAsTheFloorNotAsAShortfall() throws
Exception {
+ // given a writer that pins the count at four (min == max) while the index
has two; the ideal is
+ // the configured floor, not a size estimate, so the 0.5 ratio means
nothing about index health
+ // (the exact 0.5 boundary itself is covered on synthetic footprints in
TestRecordIndexSizingCheck)
+ writeGlobalRecordIndex();
+ TypedProperties props = globalSizingProps(4, 4, ONE_GIB);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then it is healthy, says the ideal came from the floor, and tells the
operator why that is not a shortfall
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(),
result.getSummary());
+ assertEquals("4",
result.getEffectiveConfigs().get("effective.ideal.file.group.count"));
+ assertEquals("configured minimum (floor)",
result.getEffectiveConfigs().get("effective.ideal.source"));
+ assertTrue(result.getSummary().contains("fewer than the configured minimum
of 4"), result.getSummary());
+ }
+
+ @Test
+ public void fileGroupPastTheConfiguredMaximumSizeIsUnhealthy() throws
Exception {
+ // given a maximum file-group size far below what even a hundred-record
index occupies
+ writeGlobalRecordIndex();
+ TypedProperties props = globalSizingProps(GLOBAL_FILE_GROUPS,
GLOBAL_FILE_GROUPS, "512");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then the largest file group is called out, and the remedy is the same
re-bootstrap
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus(),
result.getSummary());
+ long largest =
Long.parseLong(result.getEffectiveConfigs().get("observed.largest.file.group.bytes"));
+ assertTrue(largest > 768, "fixture should have produced a file group past
1.5 x 512 bytes, was " + largest);
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("past
the configured maximum of 512 bytes")),
+ "expected a finding about the oversized file group, got: " +
result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains("metadata delete-record-index")),
+ "expected the re-bootstrap remedy, got: " + result.getFindings());
+ }
+
+ @Test
+ public void applyAllDefaultsEvaluatesWithoutSuppliedConfig() throws
Exception {
+ // given no writer configuration at all, but defaults explicitly allowed
+ writeGlobalRecordIndex();
+
+ // when the check runs
+ HealthCheckResult result = runCheck(new TypedProperties(), true);
+
+ // then it runs rather than skipping, and reports the stock sizing it
reasoned with
+ assertNotEquals(HealthStatus.SKIPPED, result.getStatus(),
result.getSummary());
+
assertEquals(String.valueOf(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.defaultValue()),
+
result.getEffectiveConfigs().get(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key()));
+ // A fixture-sized index against the stock 1 GiB maximum file-group size
estimates zero file
+ // groups, so the ideal is the configured minimum and carries nothing
about the table. That is
+ // not a shortfall to act on: it must not read as UNHEALTHY and must not
recommend a rebuild.
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(),
result.getSummary());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("comes
from the configured minimum")),
+ "expected the floor to be named as the reason, got: " +
result.getFindings());
+ assertTrue(result.getFindings().stream().noneMatch(f ->
f.contains("metadata delete-record-index")),
+ "a floor-bound ideal must not recommend a re-bootstrap, got: " +
result.getFindings());
+ }
+
+ @Test
+ public void legacySizingKeysAreAcceptedAsAlternatives() throws Exception {
+ // given a writer that still spells the sizing keys the pre-rename way
+ writeGlobalRecordIndex();
+ TypedProperties props = new TypedProperties();
+ props.setProperty("hoodie.metadata.record.index.min.filegroup.count", "2");
+ props.setProperty("hoodie.metadata.record.index.max.filegroup.count", "2");
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key(),
ONE_GIB);
+
props.setProperty(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key(),
"2.0");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then the old keys satisfy the requirement and their values are what the
check reasons with
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(),
result.getSummary());
+ assertEquals("2", result.getEffectiveConfigs()
+
.get(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key()));
+ }
+
+ @Test
+ public void effectiveConfigIsEchoedSoAVerdictCanBeAudited() throws Exception
{
+ // given a known sizing configuration
+ writeGlobalRecordIndex();
+ TypedProperties props = globalSizingProps(GLOBAL_FILE_GROUPS,
GLOBAL_FILE_GROUPS, ONE_GIB);
+
+ // when the check runs
+ Map<String, String> configs = runCheck(props, false).getEffectiveConfigs();
+
+ // then every supplied value, derived threshold and measurement is
reported alongside the verdict
+ assertEquals("global", configs.get("observed.index.layout"));
+ assertEquals("2",
configs.get(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key()));
+ assertEquals("2",
configs.get(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key()));
+ assertEquals(ONE_GIB,
configs.get(HoodieMetadataConfig.RECORD_INDEX_MAX_FILE_GROUP_SIZE_BYTES_PROP.key()));
+ assertEquals("2.0",
configs.get(HoodieMetadataConfig.RECORD_INDEX_GROWTH_FACTOR_PROP.key()));
+ assertEquals("48", configs.get("effective.average.record.size.bytes"));
+ assertEquals("2", configs.get("effective.ideal.file.group.count"));
+ // min == max pins the count, so this fixture's ideal is the floor; the
estimate branch is exercised elsewhere
+ assertEquals("configured minimum (floor)",
configs.get("effective.ideal.source"));
+ assertEquals("1", configs.get("effective.min.healthy.file.group.count"));
+ assertEquals(String.valueOf((long) (1.5 * 1024L * 1024 * 1024)),
configs.get("effective.max.healthy.file.group.bytes"));
+ assertTrue(configs.containsKey("observed.base.file.size.bytes"),
configs.toString());
+ assertTrue(configs.get("effective.estimate.basis").startsWith("base files")
+ || configs.get("effective.estimate.basis").startsWith("log files"),
configs.toString());
+ assertEquals("2", configs.get("observed.file.group.count"));
+ assertTrue(Long.parseLong(configs.get("observed.total.size.bytes")) > 0);
+
assertTrue(Long.parseLong(configs.get("observed.largest.file.group.bytes")) >
0);
+ assertTrue(Long.parseLong(configs.get("observed.estimated.record.count"))
> 0);
+ assertTrue(Integer.parseInt(configs.get("observed.base.file.count"))
+ + Integer.parseInt(configs.get("observed.log.file.count")) > 0);
+ assertTrue(!configs.get("observed.largest.file.group.id").isEmpty());
+ }
+
+ @Test
+ public void partitionedConfigOnAGlobalIndexAsksForTheGlobalSizingKeys()
throws Exception {
+ // given a global index, but a writer configuration that says the index is
partitioned, with a
+ // maximum file-group size the index is already past
+ writeGlobalRecordIndex();
+ TypedProperties props = partitionedSizingProps(10, 10, "512");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then the layout comes from storage, so the check asks for the sizing
keys the index on disk
+ // actually needs rather than silently judging it against the partitioned
ones it was handed
+ assertEquals(HealthStatus.SKIPPED, result.getStatus(),
result.getSummary());
+ assertTrue(result.getSummary().contains(
+
HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key()),
+ "expected the global sizing key to be named, got: " +
result.getSummary());
+ // and it judges nothing until it has them -- naming the missing key beats
guessing with the
+ // wrong one, per the checker's fail-fast contract
+ assertTrue(result.getFindings().isEmpty(),
+ "a skipped check must not carry findings, got: " +
result.getFindings());
+ }
+
+ @Test
+ public void
partitionedConfigOnAGlobalIndexNeverComparesPerPartitionAgainstTheTotal()
throws Exception {
+ // given the same mismatch but a generous maximum file-group size
+ writeGlobalRecordIndex();
+ TypedProperties props = partitionedSizingProps(10, 10, ONE_GIB);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then no false shortfall: a per-partition count is never compared
against the whole-index
+ // total, because the global sizing keys govern an index that is global on
storage
+ assertEquals(HealthStatus.SKIPPED, result.getStatus(),
result.getSummary());
+ assertTrue(result.getSummary().contains(
+
HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP.key()),
+ "expected the global sizing key to be named, got: " +
result.getSummary());
+ }
+
+ @Test
+ public void partitionedIndexIsComparedPerDataPartition() throws Exception {
+ // given a partitioned index with one file group in each of the
generator's three data partitions,
+ // and a writer that still sizes it at one file group per data partition
+ writePartitionedRecordIndex();
+ TypedProperties props =
partitionedSizingProps(FILE_GROUPS_PER_DATA_PARTITION,
FILE_GROUPS_PER_DATA_PARTITION, ONE_GIB);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then the total is reported, the comparison is per data partition, and
the index is healthy
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(),
result.getSummary());
+ Map<String, String> configs = result.getEffectiveConfigs();
+ assertEquals("partitioned", configs.get("observed.index.layout"));
+ assertEquals(String.valueOf(DATA_PARTITIONS),
configs.get("observed.data.partition.count"));
+ assertEquals(String.valueOf(DATA_PARTITIONS *
FILE_GROUPS_PER_DATA_PARTITION), configs.get("observed.file.group.count"));
+ assertEquals("1",
configs.get("observed.worst.data.partition.file.group.count"));
+ assertEquals("1", configs.get("effective.ideal.file.group.count"));
+ // min == max per data partition, so every partition is floor-bound here;
the pick among
+ // size-derived partitions is covered on synthetic footprints in
TestRecordIndexSizingCheck
+ assertEquals("configured minimum (floor)",
configs.get("effective.ideal.source"));
+ assertEquals("0", configs.get("observed.undersized.data.partition.count"));
+ assertTrue(configs.get("effective.count.comparison").startsWith("per data
partition"), configs.toString());
+ assertTrue(configs.containsKey("observed.worst.data.partition"),
configs.toString());
+ }
+
+ @Test
+ public void
partitionedIndexUndersizedInADataPartitionIsUnhealthyAndNamesThePartition()
throws Exception {
+ // given a partitioned index with one file group per data partition, and a
maximum file-group
+ // size small enough that each partition's own base-file bytes call for
more than the one it has
+ writePartitionedRecordIndex();
+ TypedProperties props = partitionedSizingProps(1, 64, ONE_KIB);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(props, false);
+
+ // then the shortfall is judged per data partition -- three total file
groups are not weighed
+ // against four -- and the worst partition is named in the finding with
the remedy
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus(),
result.getSummary());
+ Map<String, String> configs = result.getEffectiveConfigs();
+ int partitionedIdeal =
Integer.parseInt(configs.get("effective.ideal.file.group.count"));
+ assertTrue(partitionedIdeal > 1, "the ideal must come from the estimate,
not the floor; was " + partitionedIdeal);
+ assertEquals(String.valueOf((int) Math.ceil(0.5 * partitionedIdeal)),
+ configs.get("effective.min.healthy.file.group.count"));
+ assertEquals("1",
configs.get("observed.worst.data.partition.file.group.count"));
+ assertEquals("size estimate", configs.get("effective.ideal.source"));
+
assertTrue(Integer.parseInt(configs.get("observed.undersized.data.partition.count"))
>= 1, configs.toString());
+ String worstPartition = configs.get("observed.worst.data.partition");
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains(
+ String.format("Data partition '%s' of the record index was initialized
with 1 file group(s)", worstPartition))),
+ "expected the finding to name the worst data partition, got: " +
result.getFindings());
+ // HoodieIndexer cannot rebuild a partitioned index (apache/hudi#20343),
so the remedy must say
+ // so rather than name a command that would silently produce a global
index instead
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains("issues/20343")),
+ "expected the partitioned-rebuild caveat, got: " +
result.getFindings());
+ assertTrue(result.getFindings().stream().noneMatch(f ->
f.contains("metadata delete-record-index")),
+ "the global re-bootstrap command must not be offered for a partitioned
index, got: " + result.getFindings());
+ }
+}
