codope commented on code in PR #20154: URL: https://github.com/apache/hudi/pull/20154#discussion_r4172237331
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTableSyncCheck.java: ########## @@ -0,0 +1,365 @@ +/* + * 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.engine.HoodieEngineContext; +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.model.HoodieLogFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.metadata.FileSystemBackedTableMetadata; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; + +/** + * Checks whether the metadata table is caught up with the data table, and whether the file listing + * it serves agrees with what is actually on storage. + * + * <p>Two signals. The first is sync lag: every completed data-table write is mirrored by a delta + * commit on the metadata table with the same instant time, so the latest completed metadata-table + * delta commit says how far the metadata table has been brought up to date. A data-table write + * newer than that is one the metadata table does not know about, and readers relying on the + * metadata table for listings will not see its files. + * + * <p>The second is listing consistency: for a bounded sample of partitions, the latest base files + * and log files are listed twice -- once through the metadata table and once straight from the + * file system -- and compared. A disagreement means the metadata table has drifted from storage, + * which is the one condition that justifies rebuilding it. The sample is bounded because listing a + * partition from the file system is precisely the expensive operation the metadata table exists to + * avoid; a daily health check must not turn into a full table scan. It is weighted towards the + * most recently written partitions, where drift actually originates, with a couple of the oldest + * added for coverage. For a complete comparison the finding points at + * {@code HoodieMetadataTableValidator}. + * + * <p>Engine-agnostic: both listing backends and the file-system views are built on + * {@link HoodieLocalEngineContext}, so no Spark session is needed. + */ +public class MetadataTableSyncCheck implements HealthCheck { + + static final String NAME = "mdt-sync"; + + /** + * Upper bound on partitions whose listings are compared. Each sampled partition costs one + * file-system listing, which is the operation the metadata table exists to avoid, so the sample + * is kept small enough to be cheap on a schedule while still catching drift that affects the + * table broadly rather than a single partition. + * + * <p>The sample is taken from the <em>end</em> of the sorted partition list. Drift comes from + * recent writes, cleans and rollbacks, so the partitions most likely to disagree are the ones + * most recently written -- and on the common date-partitioned table, lexical order is time + * order, so the last partitions are the newest. Sampling the first ones would compare the + * partitions least likely to have changed. + */ + static final int MAX_PARTITIONS_TO_COMPARE = 20; + + /** + * How many of the oldest partitions are added to the sample when the table has more partitions + * than {@link #MAX_PARTITIONS_TO_COMPARE}, so that drift confined to old data (a bad clean, a + * partition dropped by hand) is not invisible to a check that otherwise looks only at new data. + */ + static final int OLDEST_PARTITIONS_TO_INCLUDE = 2; + + /** + * How many differing file names to spell out per partition before truncating. Enough to identify + * the pattern (one file, one file group, everything) without flooding the report. + */ + static final int MAX_DIFFERING_FILES_TO_NAME = 5; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether the metadata table is caught up with the data table and its file listing matches storage"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) throws Exception { + Option<String> missing = context.firstMissingConfig(HoodieMetadataConfig.ENABLE.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + boolean metadataEnabled = Boolean.parseBoolean(context.getProps().getString( + HoodieMetadataConfig.ENABLE.key(), String.valueOf(HoodieMetadataConfig.ENABLE.defaultValue()))); + if (!metadataEnabled) { + return HealthCheckResult.skipped(NAME, "Metadata table is disabled for this writer (" + + HoodieMetadataConfig.ENABLE.key() + "=false), so there is nothing to compare against"); Review Comment: This skips the risky case. With the writer at metadata.enable=false and auto.delete.partitions=false, the metadata table stays in place and readers still use it. Could this warn when a metadata table exists but the writer has it disabled? ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTableSyncCheck.java: ########## @@ -0,0 +1,365 @@ +/* + * 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.engine.HoodieEngineContext; +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.model.HoodieLogFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.metadata.FileSystemBackedTableMetadata; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; + +/** + * Checks whether the metadata table is caught up with the data table, and whether the file listing + * it serves agrees with what is actually on storage. + * + * <p>Two signals. The first is sync lag: every completed data-table write is mirrored by a delta + * commit on the metadata table with the same instant time, so the latest completed metadata-table + * delta commit says how far the metadata table has been brought up to date. A data-table write + * newer than that is one the metadata table does not know about, and readers relying on the + * metadata table for listings will not see its files. + * + * <p>The second is listing consistency: for a bounded sample of partitions, the latest base files + * and log files are listed twice -- once through the metadata table and once straight from the + * file system -- and compared. A disagreement means the metadata table has drifted from storage, + * which is the one condition that justifies rebuilding it. The sample is bounded because listing a + * partition from the file system is precisely the expensive operation the metadata table exists to + * avoid; a daily health check must not turn into a full table scan. It is weighted towards the + * most recently written partitions, where drift actually originates, with a couple of the oldest + * added for coverage. For a complete comparison the finding points at + * {@code HoodieMetadataTableValidator}. + * + * <p>Engine-agnostic: both listing backends and the file-system views are built on + * {@link HoodieLocalEngineContext}, so no Spark session is needed. + */ +public class MetadataTableSyncCheck implements HealthCheck { + + static final String NAME = "mdt-sync"; + + /** + * Upper bound on partitions whose listings are compared. Each sampled partition costs one + * file-system listing, which is the operation the metadata table exists to avoid, so the sample + * is kept small enough to be cheap on a schedule while still catching drift that affects the + * table broadly rather than a single partition. + * + * <p>The sample is taken from the <em>end</em> of the sorted partition list. Drift comes from + * recent writes, cleans and rollbacks, so the partitions most likely to disagree are the ones + * most recently written -- and on the common date-partitioned table, lexical order is time + * order, so the last partitions are the newest. Sampling the first ones would compare the + * partitions least likely to have changed. + */ + static final int MAX_PARTITIONS_TO_COMPARE = 20; + + /** + * How many of the oldest partitions are added to the sample when the table has more partitions + * than {@link #MAX_PARTITIONS_TO_COMPARE}, so that drift confined to old data (a bad clean, a + * partition dropped by hand) is not invisible to a check that otherwise looks only at new data. + */ + static final int OLDEST_PARTITIONS_TO_INCLUDE = 2; + + /** + * How many differing file names to spell out per partition before truncating. Enough to identify + * the pattern (one file, one file group, everything) without flooding the report. + */ + static final int MAX_DIFFERING_FILES_TO_NAME = 5; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether the metadata table is caught up with the data table and its file listing matches storage"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) throws Exception { + Option<String> missing = context.firstMissingConfig(HoodieMetadataConfig.ENABLE.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + boolean metadataEnabled = Boolean.parseBoolean(context.getProps().getString( + HoodieMetadataConfig.ENABLE.key(), String.valueOf(HoodieMetadataConfig.ENABLE.defaultValue()))); + if (!metadataEnabled) { + return HealthCheckResult.skipped(NAME, "Metadata table is disabled for this writer (" + + HoodieMetadataConfig.ENABLE.key() + "=false), so there is nothing to compare against"); + } + + HoodieTableMetaClient metaClient = context.getMetaClient(); + Option<String> noMetadataTable = MetadataTablePresence.reasonAbsent(metaClient); + if (noMetadataTable.isPresent()) { + return HealthCheckResult.skipped(NAME, noMetadataTable.get()); + } + + HealthCheckResult.Builder result = HealthCheckResult.builder(NAME) + .config(HoodieMetadataConfig.ENABLE.key(), true) + .config("effective.partitions.sampled.max", MAX_PARTITIONS_TO_COMPARE); + + HoodieEngineContext engineContext = new HoodieLocalEngineContext(metaClient.getStorageConf()); + String basePath = metaClient.getBasePath().toString(); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder().enable(true).build(); + + try (HoodieTableMetadata metadataBacked = + new HoodieBackedTableMetadata(engineContext, metaClient.getStorage(), metadataConfig, basePath); + HoodieTableMetadata fileSystemBacked = + new FileSystemBackedTableMetadata(engineContext, metaClient.getTableConfig(), metaClient.getStorage(), basePath)) { + + SyncLag lag = measureSyncLag(metaClient, metadataBacked); + lag.report(result); + + ListingComparison listing = compareSampledPartitions(metaClient, metadataBacked, fileSystemBacked); + listing.report(result); + + return verdict(result, lag, listing); + } + } + + private HealthCheckResult verdict(HealthCheckResult.Builder result, SyncLag lag, ListingComparison listing) { + boolean behind = lag.commitsBehind > 0; + boolean inconsistent = !listing.differences.isEmpty(); + + if (!behind && !inconsistent) { + return result.status(HealthStatus.HEALTHY) + .summary(String.format("Metadata table is synced to %s and its listing matches storage in all %d sampled partition(s)", + lag.syncedInstant.orElse("(no data-table writes yet)"), listing.partitionsSampled)) + .build(); + } + + if (behind) { + result.finding(String.format( + "The metadata table is synced to %s but the data table's latest completed write is %s; %d completed " + + "data-table write(s) are missing from the metadata table. Readers that list through the " + + "metadata table will not see files from those writes.", + lag.syncedInstant.orElse("nothing"), lag.latestDataInstant.orElse("nothing"), lag.commitsBehind)); + result.finding("If a writer is running right now, this can be a write that is still being finalized: " + + "rerun once it completes. If the metadata table stays behind, look for failed or incomplete " + + "metadata-table delta commits ('metadata timeline show incomplete' in hudi-cli) and for " + + "metadata-table errors in the writer logs. Do not delete or rebuild the metadata table for " + + "lag alone; the next successful write brings it up to date."); + } + + for (PartitionDifference difference : listing.differences) { + result.finding(difference.describe()); + } + if (inconsistent) { + result.finding("The metadata table's listing has drifted from storage. Confirm the extent with the full " + + "comparison before acting: run HoodieMetadataTableValidator (spark-submit, hudi-utilities) with " + + "--validate-latest-file-slices --validate-latest-base-files. If it confirms the drift, the remedy " + + "is to rebuild the metadata table by disabling it for one write and re-enabling it; that is " + + "disruptive (a full listing and re-indexing of the table) and should be scheduled deliberately."); + } + + String summary = inconsistent + ? String.format("Metadata table listing differs from storage in %d of %d sampled partition(s)%s", + listing.differences.size(), listing.partitionsSampled, + behind ? ", and it is " + lag.commitsBehind + " write(s) behind" : "") + : String.format("Metadata table is %d completed data-table write(s) behind (synced to %s, latest write %s)", + lag.commitsBehind, lag.syncedInstant.orElse("nothing"), lag.latestDataInstant.orElse("nothing")); + return result.status(HealthStatus.UNHEALTHY).summary(summary).build(); + } + + private SyncLag measureSyncLag(HoodieTableMetaClient metaClient, HoodieTableMetadata metadataBacked) { + HoodieTimeline completedDataWrites = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(); + Option<String> latestDataInstant = completedDataWrites.lastInstant().map(HoodieInstant::requestedTime); + Option<String> syncedInstant = metadataBacked.getSyncedInstantTime(); + + int commitsBehind; + if (!latestDataInstant.isPresent()) { + commitsBehind = 0; + } else if (!syncedInstant.isPresent()) { + commitsBehind = completedDataWrites.countInstants(); + } else { + commitsBehind = completedDataWrites.findInstantsAfter(syncedInstant.get()).countInstants(); Review Comment: findInstantsAfter(synced) misses any gap below the synced instant. Checking each completed write with containsInstant would catch those. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTableSyncCheck.java: ########## @@ -0,0 +1,365 @@ +/* + * 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.engine.HoodieEngineContext; +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.model.HoodieLogFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.metadata.FileSystemBackedTableMetadata; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; + +/** + * Checks whether the metadata table is caught up with the data table, and whether the file listing + * it serves agrees with what is actually on storage. + * + * <p>Two signals. The first is sync lag: every completed data-table write is mirrored by a delta + * commit on the metadata table with the same instant time, so the latest completed metadata-table + * delta commit says how far the metadata table has been brought up to date. A data-table write + * newer than that is one the metadata table does not know about, and readers relying on the + * metadata table for listings will not see its files. + * + * <p>The second is listing consistency: for a bounded sample of partitions, the latest base files + * and log files are listed twice -- once through the metadata table and once straight from the + * file system -- and compared. A disagreement means the metadata table has drifted from storage, + * which is the one condition that justifies rebuilding it. The sample is bounded because listing a + * partition from the file system is precisely the expensive operation the metadata table exists to + * avoid; a daily health check must not turn into a full table scan. It is weighted towards the + * most recently written partitions, where drift actually originates, with a couple of the oldest + * added for coverage. For a complete comparison the finding points at + * {@code HoodieMetadataTableValidator}. + * + * <p>Engine-agnostic: both listing backends and the file-system views are built on + * {@link HoodieLocalEngineContext}, so no Spark session is needed. + */ +public class MetadataTableSyncCheck implements HealthCheck { + + static final String NAME = "mdt-sync"; + + /** + * Upper bound on partitions whose listings are compared. Each sampled partition costs one + * file-system listing, which is the operation the metadata table exists to avoid, so the sample + * is kept small enough to be cheap on a schedule while still catching drift that affects the + * table broadly rather than a single partition. + * + * <p>The sample is taken from the <em>end</em> of the sorted partition list. Drift comes from + * recent writes, cleans and rollbacks, so the partitions most likely to disagree are the ones + * most recently written -- and on the common date-partitioned table, lexical order is time + * order, so the last partitions are the newest. Sampling the first ones would compare the + * partitions least likely to have changed. + */ + static final int MAX_PARTITIONS_TO_COMPARE = 20; + + /** + * How many of the oldest partitions are added to the sample when the table has more partitions + * than {@link #MAX_PARTITIONS_TO_COMPARE}, so that drift confined to old data (a bad clean, a + * partition dropped by hand) is not invisible to a check that otherwise looks only at new data. + */ + static final int OLDEST_PARTITIONS_TO_INCLUDE = 2; + + /** + * How many differing file names to spell out per partition before truncating. Enough to identify + * the pattern (one file, one file group, everything) without flooding the report. + */ + static final int MAX_DIFFERING_FILES_TO_NAME = 5; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether the metadata table is caught up with the data table and its file listing matches storage"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) throws Exception { + Option<String> missing = context.firstMissingConfig(HoodieMetadataConfig.ENABLE.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + boolean metadataEnabled = Boolean.parseBoolean(context.getProps().getString( + HoodieMetadataConfig.ENABLE.key(), String.valueOf(HoodieMetadataConfig.ENABLE.defaultValue()))); + if (!metadataEnabled) { + return HealthCheckResult.skipped(NAME, "Metadata table is disabled for this writer (" + + HoodieMetadataConfig.ENABLE.key() + "=false), so there is nothing to compare against"); + } + + HoodieTableMetaClient metaClient = context.getMetaClient(); + Option<String> noMetadataTable = MetadataTablePresence.reasonAbsent(metaClient); + if (noMetadataTable.isPresent()) { + return HealthCheckResult.skipped(NAME, noMetadataTable.get()); + } + + HealthCheckResult.Builder result = HealthCheckResult.builder(NAME) + .config(HoodieMetadataConfig.ENABLE.key(), true) + .config("effective.partitions.sampled.max", MAX_PARTITIONS_TO_COMPARE); + + HoodieEngineContext engineContext = new HoodieLocalEngineContext(metaClient.getStorageConf()); + String basePath = metaClient.getBasePath().toString(); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder().enable(true).build(); + + try (HoodieTableMetadata metadataBacked = + new HoodieBackedTableMetadata(engineContext, metaClient.getStorage(), metadataConfig, basePath); + HoodieTableMetadata fileSystemBacked = + new FileSystemBackedTableMetadata(engineContext, metaClient.getTableConfig(), metaClient.getStorage(), basePath)) { + + SyncLag lag = measureSyncLag(metaClient, metadataBacked); + lag.report(result); + + ListingComparison listing = compareSampledPartitions(metaClient, metadataBacked, fileSystemBacked); + listing.report(result); + + return verdict(result, lag, listing); + } + } + + private HealthCheckResult verdict(HealthCheckResult.Builder result, SyncLag lag, ListingComparison listing) { + boolean behind = lag.commitsBehind > 0; + boolean inconsistent = !listing.differences.isEmpty(); + + if (!behind && !inconsistent) { + return result.status(HealthStatus.HEALTHY) + .summary(String.format("Metadata table is synced to %s and its listing matches storage in all %d sampled partition(s)", + lag.syncedInstant.orElse("(no data-table writes yet)"), listing.partitionsSampled)) + .build(); + } + + if (behind) { + result.finding(String.format( + "The metadata table is synced to %s but the data table's latest completed write is %s; %d completed " + + "data-table write(s) are missing from the metadata table. Readers that list through the " + + "metadata table will not see files from those writes.", + lag.syncedInstant.orElse("nothing"), lag.latestDataInstant.orElse("nothing"), lag.commitsBehind)); + result.finding("If a writer is running right now, this can be a write that is still being finalized: " + + "rerun once it completes. If the metadata table stays behind, look for failed or incomplete " + + "metadata-table delta commits ('metadata timeline show incomplete' in hudi-cli) and for " + + "metadata-table errors in the writer logs. Do not delete or rebuild the metadata table for " + + "lag alone; the next successful write brings it up to date."); + } + + for (PartitionDifference difference : listing.differences) { + result.finding(difference.describe()); + } + if (inconsistent) { + result.finding("The metadata table's listing has drifted from storage. Confirm the extent with the full " + + "comparison before acting: run HoodieMetadataTableValidator (spark-submit, hudi-utilities) with " + + "--validate-latest-file-slices --validate-latest-base-files. If it confirms the drift, the remedy " + + "is to rebuild the metadata table by disabling it for one write and re-enabling it; that is " + + "disruptive (a full listing and re-indexing of the table) and should be scheduled deliberately."); + } + + String summary = inconsistent + ? String.format("Metadata table listing differs from storage in %d of %d sampled partition(s)%s", + listing.differences.size(), listing.partitionsSampled, + behind ? ", and it is " + lag.commitsBehind + " write(s) behind" : "") + : String.format("Metadata table is %d completed data-table write(s) behind (synced to %s, latest write %s)", + lag.commitsBehind, lag.syncedInstant.orElse("nothing"), lag.latestDataInstant.orElse("nothing")); + return result.status(HealthStatus.UNHEALTHY).summary(summary).build(); + } + + private SyncLag measureSyncLag(HoodieTableMetaClient metaClient, HoodieTableMetadata metadataBacked) { + HoodieTimeline completedDataWrites = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(); + Option<String> latestDataInstant = completedDataWrites.lastInstant().map(HoodieInstant::requestedTime); + Option<String> syncedInstant = metadataBacked.getSyncedInstantTime(); + + int commitsBehind; + if (!latestDataInstant.isPresent()) { + commitsBehind = 0; + } else if (!syncedInstant.isPresent()) { + commitsBehind = completedDataWrites.countInstants(); + } else { + commitsBehind = completedDataWrites.findInstantsAfter(syncedInstant.get()).countInstants(); + } + return new SyncLag(latestDataInstant, syncedInstant, commitsBehind); + } + + /** + * Compares the latest base-file and log-file names per sampled partition between the two listing + * backends. Both views are opened over the same completed-commits timeline so that a write landing + * mid-check cannot show up as a difference. + */ + private ListingComparison compareSampledPartitions(HoodieTableMetaClient metaClient, + HoodieTableMetadata metadataBacked, + HoodieTableMetadata fileSystemBacked) throws IOException { + List<String> sampled = samplePartitions(metadataBacked.getAllPartitionPaths()); + + HoodieTimeline visibleTimeline = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants(); + List<PartitionDifference> differences = new ArrayList<>(); + try (HoodieTableFileSystemView metadataView = new HoodieTableFileSystemView(metadataBacked, metaClient, visibleTimeline); + HoodieTableFileSystemView fileSystemView = new HoodieTableFileSystemView(fileSystemBacked, metaClient, visibleTimeline)) { + for (String partition : sampled) { + PartitionDifference difference = new PartitionDifference(partition, + latestBaseFileNames(metadataView, partition), latestBaseFileNames(fileSystemView, partition), + latestLogFileNames(metadataView, partition), latestLogFileNames(fileSystemView, partition)); + if (difference.isPresent()) { + differences.add(difference); + } + } + } + return new ListingComparison(sampled.size(), differences); + } + + /** + * The partitions to compare: all of them when there are few, otherwise the newest + * {@link #MAX_PARTITIONS_TO_COMPARE} in sorted order plus the {@link #OLDEST_PARTITIONS_TO_INCLUDE} + * oldest. Sorted, so the result is stable from one run to the next. + */ + static List<String> samplePartitions(List<String> allPartitions) { + List<String> sorted = new ArrayList<>(allPartitions); + Collections.sort(sorted); + if (sorted.size() <= MAX_PARTITIONS_TO_COMPARE) { + return sorted; + } + List<String> sampled = new ArrayList<>(sorted.subList(0, OLDEST_PARTITIONS_TO_INCLUDE)); + sampled.addAll(sorted.subList(sorted.size() - MAX_PARTITIONS_TO_COMPARE, sorted.size())); + return sampled; + } + + private static Set<String> latestBaseFileNames(HoodieTableFileSystemView view, String partition) { + return view.getLatestBaseFiles(partition).map(HoodieBaseFile::getFileName).collect(Collectors.toCollection(TreeSet::new)); + } + + private static Set<String> latestLogFileNames(HoodieTableFileSystemView view, String partition) { + return view.getLatestFileSlices(partition).flatMap(FileSlice::getLogFiles) Review Comment: On table version 6, getLatestFileSlices doesn't filter uncommitted log files, so a streaming writer's in-flight logs look like drift. Could v6 compare base files only? ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTableSyncCheck.java: ########## @@ -0,0 +1,365 @@ +/* + * 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.engine.HoodieEngineContext; +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.model.HoodieLogFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.metadata.FileSystemBackedTableMetadata; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; + +/** + * Checks whether the metadata table is caught up with the data table, and whether the file listing + * it serves agrees with what is actually on storage. + * + * <p>Two signals. The first is sync lag: every completed data-table write is mirrored by a delta + * commit on the metadata table with the same instant time, so the latest completed metadata-table + * delta commit says how far the metadata table has been brought up to date. A data-table write + * newer than that is one the metadata table does not know about, and readers relying on the + * metadata table for listings will not see its files. + * + * <p>The second is listing consistency: for a bounded sample of partitions, the latest base files + * and log files are listed twice -- once through the metadata table and once straight from the + * file system -- and compared. A disagreement means the metadata table has drifted from storage, + * which is the one condition that justifies rebuilding it. The sample is bounded because listing a + * partition from the file system is precisely the expensive operation the metadata table exists to + * avoid; a daily health check must not turn into a full table scan. It is weighted towards the + * most recently written partitions, where drift actually originates, with a couple of the oldest + * added for coverage. For a complete comparison the finding points at + * {@code HoodieMetadataTableValidator}. + * + * <p>Engine-agnostic: both listing backends and the file-system views are built on + * {@link HoodieLocalEngineContext}, so no Spark session is needed. + */ +public class MetadataTableSyncCheck implements HealthCheck { + + static final String NAME = "mdt-sync"; + + /** + * Upper bound on partitions whose listings are compared. Each sampled partition costs one + * file-system listing, which is the operation the metadata table exists to avoid, so the sample + * is kept small enough to be cheap on a schedule while still catching drift that affects the + * table broadly rather than a single partition. + * + * <p>The sample is taken from the <em>end</em> of the sorted partition list. Drift comes from + * recent writes, cleans and rollbacks, so the partitions most likely to disagree are the ones + * most recently written -- and on the common date-partitioned table, lexical order is time + * order, so the last partitions are the newest. Sampling the first ones would compare the + * partitions least likely to have changed. + */ + static final int MAX_PARTITIONS_TO_COMPARE = 20; + + /** + * How many of the oldest partitions are added to the sample when the table has more partitions + * than {@link #MAX_PARTITIONS_TO_COMPARE}, so that drift confined to old data (a bad clean, a + * partition dropped by hand) is not invisible to a check that otherwise looks only at new data. + */ + static final int OLDEST_PARTITIONS_TO_INCLUDE = 2; + + /** + * How many differing file names to spell out per partition before truncating. Enough to identify + * the pattern (one file, one file group, everything) without flooding the report. + */ + static final int MAX_DIFFERING_FILES_TO_NAME = 5; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether the metadata table is caught up with the data table and its file listing matches storage"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) throws Exception { + Option<String> missing = context.firstMissingConfig(HoodieMetadataConfig.ENABLE.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + boolean metadataEnabled = Boolean.parseBoolean(context.getProps().getString( + HoodieMetadataConfig.ENABLE.key(), String.valueOf(HoodieMetadataConfig.ENABLE.defaultValue()))); + if (!metadataEnabled) { + return HealthCheckResult.skipped(NAME, "Metadata table is disabled for this writer (" + + HoodieMetadataConfig.ENABLE.key() + "=false), so there is nothing to compare against"); + } + + HoodieTableMetaClient metaClient = context.getMetaClient(); + Option<String> noMetadataTable = MetadataTablePresence.reasonAbsent(metaClient); + if (noMetadataTable.isPresent()) { + return HealthCheckResult.skipped(NAME, noMetadataTable.get()); + } + + HealthCheckResult.Builder result = HealthCheckResult.builder(NAME) + .config(HoodieMetadataConfig.ENABLE.key(), true) + .config("effective.partitions.sampled.max", MAX_PARTITIONS_TO_COMPARE); + + HoodieEngineContext engineContext = new HoodieLocalEngineContext(metaClient.getStorageConf()); + String basePath = metaClient.getBasePath().toString(); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder().enable(true).build(); + + try (HoodieTableMetadata metadataBacked = + new HoodieBackedTableMetadata(engineContext, metaClient.getStorage(), metadataConfig, basePath); + HoodieTableMetadata fileSystemBacked = + new FileSystemBackedTableMetadata(engineContext, metaClient.getTableConfig(), metaClient.getStorage(), basePath)) { + + SyncLag lag = measureSyncLag(metaClient, metadataBacked); + lag.report(result); + + ListingComparison listing = compareSampledPartitions(metaClient, metadataBacked, fileSystemBacked); + listing.report(result); + + return verdict(result, lag, listing); + } + } + + private HealthCheckResult verdict(HealthCheckResult.Builder result, SyncLag lag, ListingComparison listing) { + boolean behind = lag.commitsBehind > 0; + boolean inconsistent = !listing.differences.isEmpty(); + + if (!behind && !inconsistent) { + return result.status(HealthStatus.HEALTHY) + .summary(String.format("Metadata table is synced to %s and its listing matches storage in all %d sampled partition(s)", + lag.syncedInstant.orElse("(no data-table writes yet)"), listing.partitionsSampled)) + .build(); + } + + if (behind) { + result.finding(String.format( + "The metadata table is synced to %s but the data table's latest completed write is %s; %d completed " + + "data-table write(s) are missing from the metadata table. Readers that list through the " + + "metadata table will not see files from those writes.", + lag.syncedInstant.orElse("nothing"), lag.latestDataInstant.orElse("nothing"), lag.commitsBehind)); + result.finding("If a writer is running right now, this can be a write that is still being finalized: " + + "rerun once it completes. If the metadata table stays behind, look for failed or incomplete " + + "metadata-table delta commits ('metadata timeline show incomplete' in hudi-cli) and for " + + "metadata-table errors in the writer logs. Do not delete or rebuild the metadata table for " + + "lag alone; the next successful write brings it up to date."); + } + + for (PartitionDifference difference : listing.differences) { + result.finding(difference.describe()); + } + if (inconsistent) { + result.finding("The metadata table's listing has drifted from storage. Confirm the extent with the full " + + "comparison before acting: run HoodieMetadataTableValidator (spark-submit, hudi-utilities) with " + + "--validate-latest-file-slices --validate-latest-base-files. If it confirms the drift, the remedy " + + "is to rebuild the metadata table by disabling it for one write and re-enabling it; that is " Review Comment: Disabling the metadata table for one write only deletes it when auto.delete.partitions=true. Otherwise it adds another missed write and rebuilds nothing. It also drops every index. Could the remedy be metadata delete followed by metadata create in hudi-cli? ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataCompactionLagCheck.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.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.timeline.TimelineUtils; +import org.apache.hudi.common.util.CompactionUtils; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.storage.StoragePath; + +import java.util.Date; +import java.util.concurrent.TimeUnit; + +/** + * Checks whether compaction on the metadata table is keeping pace with the delta commits landing + * on it. + * + * <p>The metadata table is a Merge-on-Read table of its own under {@code .hoodie/metadata}, and + * every data-table write adds a delta commit to it. Its compaction is not a separately scheduled + * service: the data-table writer runs it inline, after its own commit, when the configured number + * of delta commits has accumulated. When that compaction stops running, every metadata-table read + * has to merge an ever-growing chain of log files, and the data table's own archival -- which + * cannot move past the last metadata-table compaction -- stalls with it. + * + * <p>The most common reason it stops is not a failure but the rule the writer applies to pick the + * compaction instant. To keep a compacted base file consistent with the data table, the writer + * pins the compaction instant just below the earliest data-table instant that is still pending + * but whose delta commit has already been applied to the metadata table. Once a compaction at + * that pinned instant exists, every later attempt lands on the same instant and is skipped, until + * the pending data-table instant completes or is rolled back. A long-running or abandoned + * data-table instant therefore freezes metadata-table compaction, and it is the first thing to + * look at. + * + * <p>The same slack factor as the data-table compaction check is applied before calling the table + * unhealthy: compaction lands one or two writes after the trigger point, and the aim is to catch a + * compaction that has stopped, not scheduling jitter. + */ +public class MetadataCompactionLagCheck implements HealthCheck { + + static final String NAME = "mdt-compaction"; + + /** + * How far past the configured trigger point delta commits may accumulate on the metadata table + * before it is called unhealthy. Compaction is attempted after the write that crosses the + * trigger, and one more write can land while a pending data-table instant briefly pins it. + */ + static final double TRIGGER_SLACK_FACTOR = 2.0; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether compaction on the metadata table is keeping pace with its delta commits"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) throws Exception { + Option<String> missing = context.firstMissingConfig( + HoodieMetadataConfig.ENABLE.key(), HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + boolean metadataEnabled = Boolean.parseBoolean(context.getProps().getString( + HoodieMetadataConfig.ENABLE.key(), String.valueOf(HoodieMetadataConfig.ENABLE.defaultValue()))); + if (!metadataEnabled) { + return HealthCheckResult.skipped(NAME, "Metadata table is disabled for this writer (" + + HoodieMetadataConfig.ENABLE.key() + "=false), so there is no metadata table to compact"); + } + + HoodieTableMetaClient dataMetaClient = context.getMetaClient(); + Option<String> noMetadataTable = MetadataTableSyncCheck.MetadataTablePresence.reasonAbsent(dataMetaClient); + if (noMetadataTable.isPresent()) { + return HealthCheckResult.skipped(NAME, noMetadataTable.get()); + } + + int maxDeltaCommits = Integer.parseInt(context.getProps().getString( + HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key(), + String.valueOf(HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.defaultValue()))); + int threshold = (int) Math.ceil(TRIGGER_SLACK_FACTOR * maxDeltaCommits); + + HealthCheckResult.Builder result = HealthCheckResult.builder(NAME) + .config(HoodieMetadataConfig.ENABLE.key(), true) + .config(HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key(), maxDeltaCommits) + .config("effective.unhealthy.threshold.delta.commits", threshold); + + HoodieActiveTimeline metadataTimeline = metadataTableMetaClient(dataMetaClient).getActiveTimeline(); + Option<Pair<HoodieTimeline, HoodieInstant>> sinceLastCompaction = + CompactionUtils.getCompletedDeltaCommitsSinceLatestCompaction(metadataTimeline); + if (!sinceLastCompaction.isPresent()) { + return result.status(HealthStatus.HEALTHY) + .summary("No delta commits have landed on the metadata table yet; nothing for compaction to fall behind on") + .build(); + } + + // When no compaction has completed, the instant returned is the first delta commit, not a compaction. + Option<HoodieInstant> lastCompaction = metadataTimeline.getCommitTimeline().filterCompletedInstants().lastInstant(); + HoodieTimeline deltaCommitsSince = sinceLastCompaction.get().getLeft(); + String lagStart = lagStartTime(lastCompaction, deltaCommitsSince); + Option<HoodieInstant> latestDeltaCommit = deltaCommitsSince.lastInstant(); + int deltaCommitsSinceCompaction = deltaCommitsSince.countInstants(); + Option<Long> lagMinutes = latestDeltaCommit.flatMap(latest -> minutesBetween(lagStart, latest.requestedTime())); + int pendingMetadataCompactions = metadataTimeline.filterPendingCompactionTimeline().countInstants(); + Option<HoodieInstant> earliestPendingDataInstant = dataMetaClient.getActiveTimeline().filterInflightsAndRequested().firstInstant(); Review Comment: This is right for table version 8. On v6, any pending instant blocks compaction, and the writer fails after hoodie.metadata.max.deltacommits.when_pending, so the finding should branch on table version. -- 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]
