You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2020/12/16 21:21:21 UTC
[GitHub] [hudi] rmpifer opened a new pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
rmpifer opened a new pull request #2342:
URL: https://github.com/apache/hudi/pull/2342
## *Tips*
- *Thank you very much for contributing to Apache Hudi.*
- *Please review https://hudi.apache.org/contributing.html before opening a pull request.*
## What is the purpose of the pull request
Currently every time an operation (i.e. commit, clean, rollback) completes we attempt to keep the metadata table up to date by creating an upsert on the metadata table with the same instant time. This way we can reference both datasets timelines and directly compare the instants on them to see if they are in sync.
There can be the possibility that the dataset timeline and the metadata table timeline become out of sync. When trying to read from the metadata table while the timeline is out of sync you would get incorrect values for `getAllFilesInPartition` and `getAllPartitionPaths`.
This change provides a way to overcome this scenario by reading unsynced timeline instants and merging it with existing metadata table records to get the most up to date state of the file system
JIRA: https://issues.apache.org/jira/browse/HUDI-1325
## Brief change log
#### Timeline Sync
* The logic of converting timeline metadata to metadata table records was directly tied to the commit phase in `HoodieBackedMetadataWriter`. Refactored this logic to a utility class `HoodieTableMetadataTimelineUtil`
* Created a scanner `HoodieMetadataMergedInstantRecordScanner` which handles conversion of timeline instants to metadata records and merges results
* Added final step in `AbstractHoodieTableMetadata.getMergedRecordByKey` which uses the new scanner mentioned to fetch the `HoodieRecord` associated with the desired `key` from the unsynced timeline instants and merge it with the record from the metadata table
* When converting rollback operation to metadata table records there was logic that re-read from the metadata table to ensure any files being deleted as part of roll back existed.
```
// Rollbacks deletes instants from timeline. The instant being rolled-back may not have been synced to the
// metadata table. Hence, the deleted filed need to be checked against the metadata.
```
This doesn't make sense since all instants are processed in serial order so there would never be the case where a rollback was being written before an instant earlier on the timeline was already synced. Removed this logic because it created circular dependency when implementing timeline merging
* Changed the validate metadata step in tests to use the metadata reader `HoodieBackedTableMetadata`. By default when metadata writer `HoodieBackedTableMetadataWriter` is initialized it syncs all instants to the metadata table. By using the reader we can simulate metadata table being out of sync.
* Modified `initMetaClient` in test base class to allow table type to be passed in since table type is always set as `COPY_ON_WRITE` if using this method to initialize the meta client
#### Refactor
For the following reasons I modified the HoodieTableMetadata interface to be an AbstractHoodieMetadata class which contains the following shared functionality irresepective of how the metadata is stored (as a Hoodie table, in some key/value store):
* Fetching `getAllPartitions` and `getAllFileInPartition` from metadata should validate metadata flag is enabled and should default to file listing if any error occurs regardless of storage type
* During `fetchAllPartitions` and `fetchAllFilesInPartition` metrics should be published on operations and if `validateLookUps` is enabled, results returned from metadata should be compared against actual file listing results regardless of what storage type is
* In `getMergedRecordByKey` regardless of how the key is fetched from storage the result is merged against unsynced timeline instants. This is why I introduced abstract `getRecordByKeyFromMetadata` which can be implemented by inheriting class which contains logic specific to storage type.
* Moved `findInstantsToSync` to `AbstractHoodieTableMetadata`. This method is needed regardless of storage type to find if metadata is in sync with timeline. However since how the last synced instant is stored can be dependent on storage type made this method `abstract`.
## Verify this pull request
*(Please pick either of the following options)*
This pull request is a trivial rework / code cleanup without any test coverage.
*(or)*
This pull request is already covered by existing tests, such as *(please describe tests)*.
(or)
This change added tests and can be verified as follows:
### Testing
Ran `TestHoodieBackedMetadata` which contains all tests related to metadata table
Added tests which use an unsynced client after commits, cleans, and restores have been performed to ensure updates not written to the metadata yet are still being reflected when reading from the metadata
## Committer checklist
- [ ] Has a corresponding JIRA in PR title & commit
- [ ] Commit message is descriptive of the change
- [ ] CI is green
- [ ] Necessary doc changes done or have another open PR
- [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r547150618
##########
File path: hudi-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java
##########
@@ -619,32 +618,97 @@ public void testMetadataMetrics() throws Exception {
}
}
+ //@ParameterizedTest
+ //@EnumSource(HoodieTableType.class)
+ public void testMetadataOutOfSync(HoodieTableType tableType) throws Exception {
+ init(HoodieTableType.COPY_ON_WRITE);
+
+ HoodieWriteClient unsyncedClient = new HoodieWriteClient<>(jsc, getWriteConfig(true, true));
+
+ // Enable metadata so table is initialized
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, true))) {
+ // Perform Bulk Insert
+ String newCommitTime = "001";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.bulkInsert(jsc.parallelize(records, 1), newCommitTime).collect();
+ }
+
+ // Perform commit operations with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ // Perform Insert
+ String newCommitTime = "002";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.insert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Perform Upsert
+ newCommitTime = "003";
+ client.startCommitWithTime(newCommitTime);
+ records = dataGen.generateUniqueUpdates(newCommitTime, 20);
+ client.upsert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Compaction
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ newCommitTime = "004";
+ client.scheduleCompactionAtInstant(newCommitTime, Option.empty());
+ client.compact(newCommitTime);
+ }
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
Review comment:
Got it. Makes sense
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-748554044
@rmpifer On this point specifically,
> so there would never be the case where a rollback was being written before an instant earlier on the timeline was already synced. Removed this logic because it created circular dependency when implementing timeline merging
Following situation is technically possible. Timeline has commits c1, c2, c3, c4 and we have synced only upto C2. User decides to restore the table to commit time c2, thus generating rollback metadata for c3, c4 which now have no files in the metadata table. I think we should defensively handle this scenario at the payload merge level i.e if there is a delete for a file that does not exist in the base file or prior instants, we should just do a no-op and continue. What I am saying is : I agree with you that the extra logic need not be present, but we should check the payload once to ensure we can handle the scenario above.
Let me know if we are not talking about the same thing :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r547154542
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/AbstractHoodieTableMetadata.java
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.avro.model.HoodieMetadataRecord;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.model.HoodiePartitionMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Interface that supports querying various pieces of metadata about a hudi table.
+ */
+public abstract class AbstractHoodieTableMetadata implements HoodieTableMetadata {
+
+ private static final Logger LOG = LogManager.getLogger(AbstractHoodieTableMetadata.class);
+
+ static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
+ static final int BUFFER_SIZE = 10 * 1024 * 1024;
+
+ protected final SerializableConfiguration hadoopConf;
+ protected final Option<HoodieMetadataMetrics> metrics;
+ protected final String datasetBasePath;
+
+ // Directory used for Spillable Map when merging records
+ final String spillableMapDirectory;
+
+ protected boolean enabled;
+ private final boolean validateLookups;
+ private final boolean assumeDatePartitioning;
+
+ private transient HoodieMetadataMergedInstantRecordScanner timelineRecordScanner;
+
+ protected AbstractHoodieTableMetadata(Configuration hadoopConf, String datasetBasePath, String spillableMapDirectory,
+ boolean enabled, boolean validateLookups, boolean enableMetrics,
+ boolean assumeDatePartitioning) {
+ this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this.datasetBasePath = datasetBasePath;
+ this.spillableMapDirectory = spillableMapDirectory;
+
+ this.enabled = enabled;
+ this.validateLookups = validateLookups;
+ this.assumeDatePartitioning = assumeDatePartitioning;
+
+ if (enableMetrics) {
+ this.metrics = Option.of(new HoodieMetadataMetrics(Registry.getRegistry("HoodieMetadata")));
+ } else {
+ this.metrics = Option.empty();
+ }
+ }
+
+ public static AbstractHoodieTableMetadata create(Configuration conf, String datasetBasePath, String spillableMapPath, boolean useFileListingFromMetadata,
+ boolean verifyListings, boolean enableMetrics, boolean shouldAssumeDatePartitioning) {
+ return new HoodieBackedTableMetadata(conf, datasetBasePath, spillableMapPath, useFileListingFromMetadata, verifyListings,
+ enableMetrics, shouldAssumeDatePartitioning);
+ }
+
+
+ /**
+ * Return the list of files in a partition.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ * @param partitionPath The absolute path of the partition to list
+ */
+
+ public FileStatus[] getAllFilesInPartition(Path partitionPath) throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllFilesInPartition(partitionPath);
+ } catch (Exception e) {
+ LOG.error("Failed to retrive files in partition " + partitionPath + " from metadata", e);
+ }
+ }
+ return FSUtils.getFs(partitionPath.toString(), hadoopConf.get()).listStatus(partitionPath);
+ }
+
+ /**
+ * Return the list of partitions in the dataset.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ */
+ public List<String> getAllPartitionPaths() throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllPartitionPaths();
+ } catch (Exception e) {
+ LOG.error("Failed to retrieve list of partition from metadata", e);
+ }
+ }
+
+ FileSystem fs = FSUtils.getFs(datasetBasePath, hadoopConf.get());
+ return FSUtils.getAllPartitionPaths(fs, datasetBasePath, assumeDatePartitioning);
+ }
+
+ /**
+ * Returns a list of all partitions.
+ */
+ /**
+ * Returns a list of all partitions.
+ */
+ protected List<String> fetchAllPartitionPaths() throws IOException {
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(RECORDKEY_PARTITION_LIST);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_PARTITIONS_STR, timer.endTimer()));
+
+ List<String> partitions = Collections.emptyList();
+ if (hoodieRecord.isPresent()) {
+ if (!hoodieRecord.get().getData().getDeletions().isEmpty()) {
+ throw new HoodieMetadataException("Metadata partition list record is inconsistent: "
+ + hoodieRecord.get().getData());
+ }
+
+ partitions = hoodieRecord.get().getData().getFilenames();
+ // Partition-less tables have a single empty partition
+ if (partitions.contains(NON_PARTITIONED_NAME)) {
+ partitions.remove(NON_PARTITIONED_NAME);
+ partitions.add("");
+ }
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<String> actualPartitions = FSUtils.getAllPartitionPaths(metaClient.getFs(), datasetBasePath, false);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_PARTITIONS_STR, timer.endTimer()));
+
+ Collections.sort(actualPartitions);
+ Collections.sort(partitions);
+ if (!actualPartitions.equals(partitions)) {
+ LOG.error("Validation of metadata partition list failed. Lists do not match.");
+ LOG.error("Partitions from metadata: " + Arrays.toString(partitions.toArray()));
+ LOG.error("Partitions from file system: " + Arrays.toString(actualPartitions.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ partitions = actualPartitions;
+ }
+
+ LOG.info("Listed partitions from metadata: #partitions=" + partitions.size());
+ return partitions;
+ }
+
+ /**
+ * Return all the files from the partition.
+ *
+ * @param partitionPath The absolute path of the partition
+ */
+ private FileStatus[] fetchAllFilesInPartition(Path partitionPath) throws IOException {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(datasetBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ partitionName = NON_PARTITIONED_NAME;
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(partitionName);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+
+ FileStatus[] statuses = {};
+ if (hoodieRecord.isPresent()) {
+ statuses = hoodieRecord.get().getData().getFileStatuses(partitionPath);
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+
+ // Ignore partition metadata file
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ FileStatus[] directStatuses = metaClient.getFs().listStatus(partitionPath,
+ p -> !p.getName().equals(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE));
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_FILES_STR, timer.endTimer()));
+
+ List<String> directFilenames = Arrays.stream(directStatuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ List<String> metadataFilenames = Arrays.stream(statuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ if (!metadataFilenames.equals(directFilenames)) {
+ LOG.error("Validation of metadata file listing for partition " + partitionName + " failed.");
+ LOG.error("File list from metadata: " + Arrays.toString(metadataFilenames.toArray()));
+ LOG.error("File list from direct listing: " + Arrays.toString(directFilenames.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ statuses = directStatuses;
+ }
+
+ LOG.info("Listed file in partition from metadata: partition=" + partitionName + ", #files=" + statuses.length);
+ return statuses;
+ }
+
+ /**
+ * Retrieve the merged {@code HoodieRecord} mapped to the given key.
+ *
+ * @param key The key of the record
+ */
+ private Option<HoodieRecord<HoodieMetadataPayload>> getMergedRecordByKey(String key) throws IOException {
+ openTimelineScanner();
+
+ Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
+
+ // Retrieve record from unsynced timeline instants
+ Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
+ if (timelineHoodieRecord.isPresent()) {
+ if (metadataHoodieRecord.isPresent()) {
+ HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
+ metadataHoodieRecord = Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
+ } else {
+ metadataHoodieRecord = timelineHoodieRecord;
Review comment:
Sorry, my core concern was that we are reassigning to a variable that holds the returned value from a key method call `getRecordByKeyFromMetadata`
I was suggesting something like.
```
Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
// Retrieve record from unsynced timeline instants
Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
if (timelineHoodieRecord.isPresent()) {
if (metadataHoodieRecord.isPresent()) {
HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
return Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
} else {
return timelineHoodieRecord;
}
}
return metadataHoodieRecord;
```
if you feel strongly, we can keep as-is.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer edited a comment on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer edited a comment on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-750456129
> ```
> data timeline : C1, rollback C12, rollback C11, ..... rollback C2
> metadata timeline: C3, ..., C8
> ```
>
> It's now hard to determine whether or not `rollback C2` should be applied. (we could compare the latest instant on metadata timeline C8 and assume its applied?)
Regarding this scenario, my thoughts would be to check rollback metadata only against unsynced instants, i.e. if any instants after C8 were considered in rollback. Is this potentially surfacing a more general issue though. Say metadata timeline has been synced up to C2, but data timeline archival has happened to C8. It seems next time we try to find unsynced instants we won't consider anything between C2...C8
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-750473227
>Regarding this scenario, my thoughts would be to check rollback metadata only against unsynced instants, i.e. if any instants after C8 were considered in rollback.
To be clear, we want to skip syncing a rollback instant to metadata table, if the instant it rolled back is not in completed state on the unsynced portion of the data timeline.? Thats what I meant by `(we could compare the latest instant on metadata timeline C8 and assume its applied?)`. I think it will work out fine. Let's implement this?
>Say metadata timeline has been synced up to C2, but data timeline archival has happened to C8.
We have code to guard against this already. Data timeline archiving won't move past latest instant on metadata timeline
https://github.com/apache/hudi/blob/f7f8c33f804b1bb2cfecf3ac18edc9eb584c679e/hudi-client/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java#L203
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar merged pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar merged pull request #2342:
URL: https://github.com/apache/hudi/pull/2342
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r545394203
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataTimelineUtil.java
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.metadata;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+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.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CleanerUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.metadata.AbstractHoodieTableMetadata.NON_PARTITIONED_NAME;
+
+/**
+ * A utility to convert timeline information to metadata table records.
+ */
+public class HoodieTableMetadataTimelineUtil {
Review comment:
I think we can simply call this `HoodieTableMetadataUtils` ?
##########
File path: hudi-client/src/main/java/org/apache/hudi/table/HoodieTable.java
##########
@@ -635,9 +635,9 @@ public boolean requireSortedRecords() {
return getBaseFileFormat() == HoodieFileFormat.HFILE;
}
- public HoodieTableMetadata metadata() {
+ public AbstractHoodieTableMetadata metadata() {
Review comment:
I would prefer to still have an interface for `HoodieTableMetadata` and return that here. We can have the abstract class internal to the actual implementation?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-751794768
@rmpifer no worries. will review today and land
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-747486716
@rmpifer could you please rebase this against latest `rfc-15` branch. I ll get started with the review in the meantime
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-750994186
@rmpifer @prashantwason lets just implement the check as discussed above and move on. We can file a follow-on JIRA if needed.
I would like merge the feature branch to master asap, to give sometime for testing
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] prashantwason commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
prashantwason commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r547076397
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -372,19 +191,6 @@ protected void closeReaders() {
logRecordScanner = null;
}
Review comment:
I think there needs to be a corresponding close() for the timelineRecordScanner too.
Once we commit an update to the table, the unsynced instants have changed.
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMergedInstantRecordScanner.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.common.util.HoodieRecordSizeEstimator;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ExternalSpillableMap;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Provides functionality to convert timeline instants to table metadata records and then merge by key. Specify
+ * a filter to limit keys that are merged and stored in memory.
+ */
+public class HoodieMetadataMergedInstantRecordScanner {
+
+ private static final Logger LOG = LogManager.getLogger(HoodieMetadataMergedInstantRecordScanner.class);
+
+ HoodieTableMetaClient metaClient;
+ private List<HoodieInstant> instants;
+ private Set<String> mergeKeyFilter;
+ protected final ExternalSpillableMap<String, HoodieRecord<? extends HoodieRecordPayload>> records;
+
+ public HoodieMetadataMergedInstantRecordScanner(HoodieTableMetaClient metaClient, List<HoodieInstant> instants,
+ Schema readerSchema, Long maxMemorySizeInBytes,
+ String spillableMapBasePath, Set<String> mergeKeyFilter) throws IOException {
+ this.metaClient = metaClient;
+ this.instants = instants;
+ this.mergeKeyFilter = mergeKeyFilter != null ? mergeKeyFilter : Collections.emptySet();
+ this.records = new ExternalSpillableMap<>(maxMemorySizeInBytes, spillableMapBasePath, new DefaultSizeEstimator(),
+ new HoodieRecordSizeEstimator(readerSchema));
+
+ scan();
+ }
+
+ private void scan() {
+ for (HoodieInstant instant : instants) {
+ try {
+ processInstant(instant);
+ } catch (Exception e) {
+ LOG.error(String.format("Got exception when processing timeline instant %s", instant.getTimestamp()), e);
+ throw new HoodieException(String.format("Got exception when processing timeline instant %s", instant.getTimestamp()), e);
+ }
+ }
+ }
+
+ /**
+ * Converts an instant to metadata table records and processes each record.
+ *
+ * @param instant
+ * @throws IOException
+ */
+ private void processInstant(HoodieInstant instant) throws IOException {
Review comment:
can be merged with the above function. I dont see it being called anywhere else.
##########
File path: hudi-client/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
##########
@@ -266,7 +256,7 @@ private void initialize(JavaSparkContext jsc, HoodieTableMetaClient datasetMetaC
}
private void initTableMetadata() {
- this.metadata = new HoodieBackedTableMetadata(hadoopConf.get(), datasetWriteConfig.getBasePath(), datasetWriteConfig.getSpillableMapBasePath(),
Review comment:
The reader-write pairs in metadata are related. So this can be simplified to be simply new HoodieBackedTableMetadata.
HoodieBackedTableMetadata cannot work with any other implementaton of HoodieTableMetadata.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer edited a comment on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer edited a comment on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-751585060
Apologies for the delay in pushing changes, was offline over the holidays :). Updated to address feedback and to resolve our discussions to check whether the commit being rolled back is after last sync point on metadata timeline
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-749435267
>If this is the case what happens when c3 and c4 are then finally synced? These files would remain in the metadata table even though rollback should have removed them
So restore in that example, would delete C3, C4 from the timeline. it won't be synced at all.
>How do we differentiate this from a bug which lead to files missing from the metadata listing? The current merge code provides some sort of correctness - deletes are always for files already created. I think we should maintain this.
I understand the need for wanting to fail-fast. but the scenario above is legit, right? Do you suggest we sync the rollbacks only if the original instants for which the rollback was done were synced to the metadata table? cc @prashantwason
Let's make the call on this one. Otherwise PR seems ready more or less, after another rebase/set of comments. ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-749988417
>Do you suggest we sync the rollbacks only if the original instants for which the rollback was done were synced to the metadata table?
Thinking out loud whether there are tricky cases here. Specifically, is there a chance of skipping a rollback instant incorrectly.
Let's say we restored the last 11 commits (12 -> 2) with metadata synced upto C8
```
data timeline : C1, rollback C12, rollback C11, ..... rollback C2
metadata timeline : C1, C2, ... C8
```
data timeline archival does not move past C8, but can technically archive C8. Similarly metadata timeline is free to archive independently and lets say it archives upto C2. If it ends up looking like
```
data timeline : C1, rollback C12, rollback C11, ..... rollback C2
metadata timeline: C3, ..., C8
```
It's now hard to determine whether or not `rollback C2` should be applied. (we could compare the latest instant on metadata timeline C8 and assume its applied?)
I think this complexity is unwarranted for sake of failing-fast. We are better of defensively, ignoring deletes for files that were not logged before.
@prashantwason Hope you can be convinced of this :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r546930182
##########
File path: hudi-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java
##########
@@ -619,32 +618,97 @@ public void testMetadataMetrics() throws Exception {
}
}
+ //@ParameterizedTest
+ //@EnumSource(HoodieTableType.class)
+ public void testMetadataOutOfSync(HoodieTableType tableType) throws Exception {
+ init(HoodieTableType.COPY_ON_WRITE);
+
+ HoodieWriteClient unsyncedClient = new HoodieWriteClient<>(jsc, getWriteConfig(true, true));
+
+ // Enable metadata so table is initialized
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, true))) {
+ // Perform Bulk Insert
+ String newCommitTime = "001";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.bulkInsert(jsc.parallelize(records, 1), newCommitTime).collect();
+ }
+
+ // Perform commit operations with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ // Perform Insert
+ String newCommitTime = "002";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.insert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Perform Upsert
+ newCommitTime = "003";
+ client.startCommitWithTime(newCommitTime);
+ records = dataGen.generateUniqueUpdates(newCommitTime, 20);
+ client.upsert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Compaction
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ newCommitTime = "004";
+ client.scheduleCompactionAtInstant(newCommitTime, Option.empty());
+ client.compact(newCommitTime);
+ }
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
+ validateMetadata(unsyncedClient);
+
+ // Perform clean operation with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ // One more commit needed to trigger clean so upsert and compact
+ String newCommitTime = "005";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateUpdates(newCommitTime, 20);
+ client.upsert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ newCommitTime = "006";
+ client.scheduleCompactionAtInstant(newCommitTime, Option.empty());
+ client.compact(newCommitTime);
+ }
+
+ // Clean
+ newCommitTime = "007";
+ client.clean(newCommitTime);
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
+ validateMetadata(unsyncedClient);
+
+ // Perform restore with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ client.restoreToInstant("004");
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
+ validateMetadata(unsyncedClient);
+ }
+
/**
* Validate the metadata tables contents to ensure it matches what is on the file system.
*
* @throws IOException
*/
private void validateMetadata(HoodieWriteClient client) throws IOException {
HoodieWriteConfig config = client.getConfig();
- HoodieBackedTableMetadataWriter metadataWriter = metadataWriter(client);
- assertNotNull(metadataWriter, "MetadataWriter should have been initialized");
+
+ HoodieBackedTableMetadata metadataReader = metadata(client);
Review comment:
Updated
##########
File path: hudi-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java
##########
@@ -619,32 +618,97 @@ public void testMetadataMetrics() throws Exception {
}
}
+ //@ParameterizedTest
+ //@EnumSource(HoodieTableType.class)
+ public void testMetadataOutOfSync(HoodieTableType tableType) throws Exception {
+ init(HoodieTableType.COPY_ON_WRITE);
+
+ HoodieWriteClient unsyncedClient = new HoodieWriteClient<>(jsc, getWriteConfig(true, true));
+
+ // Enable metadata so table is initialized
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, true))) {
+ // Perform Bulk Insert
+ String newCommitTime = "001";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.bulkInsert(jsc.parallelize(records, 1), newCommitTime).collect();
+ }
+
+ // Perform commit operations with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ // Perform Insert
+ String newCommitTime = "002";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.insert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Perform Upsert
+ newCommitTime = "003";
+ client.startCommitWithTime(newCommitTime);
+ records = dataGen.generateUniqueUpdates(newCommitTime, 20);
+ client.upsert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Compaction
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ newCommitTime = "004";
+ client.scheduleCompactionAtInstant(newCommitTime, Option.empty());
+ client.compact(newCommitTime);
+ }
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
Review comment:
I disabled metadata in the write client used for these commit operations so it would not be synced. This way we could simulate timeline and metadata being out of sync.
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -66,24 +58,13 @@
* If the metadata table does not exist, RPC calls are used to retrieve file listings from the file system.
* No updates are applied to the table and it is not synced.
*/
-public class HoodieBackedTableMetadata implements HoodieTableMetadata {
+public class HoodieBackedTableMetadata extends AbstractHoodieTableMetadata {
Review comment:
There still needs to be a place where the instant scanner result is merged with the metadata table result. I'm not sure I'm able to see how we would like this logic at scanner level. Since this should be common behavior regardless of metadata storage this is why I thought to create this
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -117,172 +93,15 @@ public HoodieBackedTableMetadata(Configuration conf, String datasetBasePath, Str
} else {
LOG.info("Metadata table is disabled.");
}
-
Review comment:
Correct
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/AbstractHoodieTableMetadata.java
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.avro.model.HoodieMetadataRecord;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.model.HoodiePartitionMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Interface that supports querying various pieces of metadata about a hudi table.
+ */
+public abstract class AbstractHoodieTableMetadata implements HoodieTableMetadata {
+
+ private static final Logger LOG = LogManager.getLogger(AbstractHoodieTableMetadata.class);
+
+ static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
+ static final int BUFFER_SIZE = 10 * 1024 * 1024;
+
+ protected final SerializableConfiguration hadoopConf;
+ protected final Option<HoodieMetadataMetrics> metrics;
+ protected final String datasetBasePath;
+
+ // Directory used for Spillable Map when merging records
+ final String spillableMapDirectory;
+
+ protected boolean enabled;
+ private final boolean validateLookups;
+ private final boolean assumeDatePartitioning;
+
+ private transient HoodieMetadataMergedInstantRecordScanner timelineRecordScanner;
+
+ protected AbstractHoodieTableMetadata(Configuration hadoopConf, String datasetBasePath, String spillableMapDirectory,
+ boolean enabled, boolean validateLookups, boolean enableMetrics,
+ boolean assumeDatePartitioning) {
+ this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this.datasetBasePath = datasetBasePath;
+ this.spillableMapDirectory = spillableMapDirectory;
+
+ this.enabled = enabled;
+ this.validateLookups = validateLookups;
+ this.assumeDatePartitioning = assumeDatePartitioning;
+
+ if (enableMetrics) {
+ this.metrics = Option.of(new HoodieMetadataMetrics(Registry.getRegistry("HoodieMetadata")));
+ } else {
+ this.metrics = Option.empty();
+ }
+ }
+
+ public static AbstractHoodieTableMetadata create(Configuration conf, String datasetBasePath, String spillableMapPath, boolean useFileListingFromMetadata,
+ boolean verifyListings, boolean enableMetrics, boolean shouldAssumeDatePartitioning) {
+ return new HoodieBackedTableMetadata(conf, datasetBasePath, spillableMapPath, useFileListingFromMetadata, verifyListings,
+ enableMetrics, shouldAssumeDatePartitioning);
+ }
+
+
+ /**
+ * Return the list of files in a partition.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ * @param partitionPath The absolute path of the partition to list
+ */
+
+ public FileStatus[] getAllFilesInPartition(Path partitionPath) throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllFilesInPartition(partitionPath);
+ } catch (Exception e) {
+ LOG.error("Failed to retrive files in partition " + partitionPath + " from metadata", e);
+ }
+ }
+ return FSUtils.getFs(partitionPath.toString(), hadoopConf.get()).listStatus(partitionPath);
+ }
+
+ /**
+ * Return the list of partitions in the dataset.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ */
+ public List<String> getAllPartitionPaths() throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllPartitionPaths();
+ } catch (Exception e) {
+ LOG.error("Failed to retrieve list of partition from metadata", e);
+ }
+ }
+
+ FileSystem fs = FSUtils.getFs(datasetBasePath, hadoopConf.get());
+ return FSUtils.getAllPartitionPaths(fs, datasetBasePath, assumeDatePartitioning);
+ }
+
+ /**
+ * Returns a list of all partitions.
+ */
+ /**
+ * Returns a list of all partitions.
+ */
+ protected List<String> fetchAllPartitionPaths() throws IOException {
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(RECORDKEY_PARTITION_LIST);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_PARTITIONS_STR, timer.endTimer()));
+
+ List<String> partitions = Collections.emptyList();
+ if (hoodieRecord.isPresent()) {
+ if (!hoodieRecord.get().getData().getDeletions().isEmpty()) {
+ throw new HoodieMetadataException("Metadata partition list record is inconsistent: "
+ + hoodieRecord.get().getData());
+ }
+
+ partitions = hoodieRecord.get().getData().getFilenames();
+ // Partition-less tables have a single empty partition
+ if (partitions.contains(NON_PARTITIONED_NAME)) {
+ partitions.remove(NON_PARTITIONED_NAME);
+ partitions.add("");
+ }
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<String> actualPartitions = FSUtils.getAllPartitionPaths(metaClient.getFs(), datasetBasePath, false);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_PARTITIONS_STR, timer.endTimer()));
+
+ Collections.sort(actualPartitions);
+ Collections.sort(partitions);
+ if (!actualPartitions.equals(partitions)) {
+ LOG.error("Validation of metadata partition list failed. Lists do not match.");
+ LOG.error("Partitions from metadata: " + Arrays.toString(partitions.toArray()));
+ LOG.error("Partitions from file system: " + Arrays.toString(actualPartitions.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ partitions = actualPartitions;
+ }
+
+ LOG.info("Listed partitions from metadata: #partitions=" + partitions.size());
+ return partitions;
+ }
+
+ /**
+ * Return all the files from the partition.
+ *
+ * @param partitionPath The absolute path of the partition
+ */
+ private FileStatus[] fetchAllFilesInPartition(Path partitionPath) throws IOException {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(datasetBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ partitionName = NON_PARTITIONED_NAME;
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(partitionName);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+
+ FileStatus[] statuses = {};
+ if (hoodieRecord.isPresent()) {
+ statuses = hoodieRecord.get().getData().getFileStatuses(partitionPath);
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+
+ // Ignore partition metadata file
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ FileStatus[] directStatuses = metaClient.getFs().listStatus(partitionPath,
+ p -> !p.getName().equals(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE));
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_FILES_STR, timer.endTimer()));
+
+ List<String> directFilenames = Arrays.stream(directStatuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ List<String> metadataFilenames = Arrays.stream(statuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ if (!metadataFilenames.equals(directFilenames)) {
+ LOG.error("Validation of metadata file listing for partition " + partitionName + " failed.");
+ LOG.error("File list from metadata: " + Arrays.toString(metadataFilenames.toArray()));
+ LOG.error("File list from direct listing: " + Arrays.toString(directFilenames.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ statuses = directStatuses;
+ }
+
+ LOG.info("Listed file in partition from metadata: partition=" + partitionName + ", #files=" + statuses.length);
+ return statuses;
+ }
+
+ /**
+ * Retrieve the merged {@code HoodieRecord} mapped to the given key.
+ *
+ * @param key The key of the record
+ */
+ private Option<HoodieRecord<HoodieMetadataPayload>> getMergedRecordByKey(String key) throws IOException {
+ openTimelineScanner();
+
+ Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
+
+ // Retrieve record from unsynced timeline instants
+ Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
+ if (timelineHoodieRecord.isPresent()) {
+ if (metadataHoodieRecord.isPresent()) {
+ HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
+ metadataHoodieRecord = Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
+ } else {
+ metadataHoodieRecord = timelineHoodieRecord;
Review comment:
I created separate variable for this so there is still one place of return
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/AbstractHoodieTableMetadata.java
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.avro.model.HoodieMetadataRecord;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.model.HoodiePartitionMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Interface that supports querying various pieces of metadata about a hudi table.
+ */
+public abstract class AbstractHoodieTableMetadata implements HoodieTableMetadata {
+
+ private static final Logger LOG = LogManager.getLogger(AbstractHoodieTableMetadata.class);
+
+ static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
+ static final int BUFFER_SIZE = 10 * 1024 * 1024;
+
+ protected final SerializableConfiguration hadoopConf;
+ protected final Option<HoodieMetadataMetrics> metrics;
+ protected final String datasetBasePath;
+
+ // Directory used for Spillable Map when merging records
+ final String spillableMapDirectory;
+
+ protected boolean enabled;
+ private final boolean validateLookups;
+ private final boolean assumeDatePartitioning;
+
+ private transient HoodieMetadataMergedInstantRecordScanner timelineRecordScanner;
+
+ protected AbstractHoodieTableMetadata(Configuration hadoopConf, String datasetBasePath, String spillableMapDirectory,
+ boolean enabled, boolean validateLookups, boolean enableMetrics,
+ boolean assumeDatePartitioning) {
+ this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this.datasetBasePath = datasetBasePath;
+ this.spillableMapDirectory = spillableMapDirectory;
+
+ this.enabled = enabled;
+ this.validateLookups = validateLookups;
+ this.assumeDatePartitioning = assumeDatePartitioning;
+
+ if (enableMetrics) {
+ this.metrics = Option.of(new HoodieMetadataMetrics(Registry.getRegistry("HoodieMetadata")));
+ } else {
+ this.metrics = Option.empty();
+ }
+ }
+
+ public static AbstractHoodieTableMetadata create(Configuration conf, String datasetBasePath, String spillableMapPath, boolean useFileListingFromMetadata,
+ boolean verifyListings, boolean enableMetrics, boolean shouldAssumeDatePartitioning) {
+ return new HoodieBackedTableMetadata(conf, datasetBasePath, spillableMapPath, useFileListingFromMetadata, verifyListings,
+ enableMetrics, shouldAssumeDatePartitioning);
+ }
+
+
+ /**
+ * Return the list of files in a partition.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ * @param partitionPath The absolute path of the partition to list
+ */
+
+ public FileStatus[] getAllFilesInPartition(Path partitionPath) throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllFilesInPartition(partitionPath);
+ } catch (Exception e) {
+ LOG.error("Failed to retrive files in partition " + partitionPath + " from metadata", e);
+ }
+ }
+ return FSUtils.getFs(partitionPath.toString(), hadoopConf.get()).listStatus(partitionPath);
+ }
+
+ /**
+ * Return the list of partitions in the dataset.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ */
+ public List<String> getAllPartitionPaths() throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllPartitionPaths();
+ } catch (Exception e) {
+ LOG.error("Failed to retrieve list of partition from metadata", e);
+ }
+ }
+
+ FileSystem fs = FSUtils.getFs(datasetBasePath, hadoopConf.get());
+ return FSUtils.getAllPartitionPaths(fs, datasetBasePath, assumeDatePartitioning);
+ }
+
+ /**
+ * Returns a list of all partitions.
+ */
+ /**
+ * Returns a list of all partitions.
+ */
+ protected List<String> fetchAllPartitionPaths() throws IOException {
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(RECORDKEY_PARTITION_LIST);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_PARTITIONS_STR, timer.endTimer()));
+
+ List<String> partitions = Collections.emptyList();
+ if (hoodieRecord.isPresent()) {
+ if (!hoodieRecord.get().getData().getDeletions().isEmpty()) {
+ throw new HoodieMetadataException("Metadata partition list record is inconsistent: "
+ + hoodieRecord.get().getData());
+ }
+
+ partitions = hoodieRecord.get().getData().getFilenames();
+ // Partition-less tables have a single empty partition
+ if (partitions.contains(NON_PARTITIONED_NAME)) {
+ partitions.remove(NON_PARTITIONED_NAME);
+ partitions.add("");
+ }
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<String> actualPartitions = FSUtils.getAllPartitionPaths(metaClient.getFs(), datasetBasePath, false);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_PARTITIONS_STR, timer.endTimer()));
+
+ Collections.sort(actualPartitions);
+ Collections.sort(partitions);
+ if (!actualPartitions.equals(partitions)) {
+ LOG.error("Validation of metadata partition list failed. Lists do not match.");
+ LOG.error("Partitions from metadata: " + Arrays.toString(partitions.toArray()));
+ LOG.error("Partitions from file system: " + Arrays.toString(actualPartitions.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ partitions = actualPartitions;
+ }
+
+ LOG.info("Listed partitions from metadata: #partitions=" + partitions.size());
+ return partitions;
+ }
+
+ /**
+ * Return all the files from the partition.
+ *
+ * @param partitionPath The absolute path of the partition
+ */
+ private FileStatus[] fetchAllFilesInPartition(Path partitionPath) throws IOException {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(datasetBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ partitionName = NON_PARTITIONED_NAME;
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(partitionName);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+
+ FileStatus[] statuses = {};
+ if (hoodieRecord.isPresent()) {
+ statuses = hoodieRecord.get().getData().getFileStatuses(partitionPath);
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+
+ // Ignore partition metadata file
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ FileStatus[] directStatuses = metaClient.getFs().listStatus(partitionPath,
+ p -> !p.getName().equals(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE));
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_FILES_STR, timer.endTimer()));
+
+ List<String> directFilenames = Arrays.stream(directStatuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ List<String> metadataFilenames = Arrays.stream(statuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ if (!metadataFilenames.equals(directFilenames)) {
+ LOG.error("Validation of metadata file listing for partition " + partitionName + " failed.");
+ LOG.error("File list from metadata: " + Arrays.toString(metadataFilenames.toArray()));
+ LOG.error("File list from direct listing: " + Arrays.toString(directFilenames.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ statuses = directStatuses;
+ }
+
+ LOG.info("Listed file in partition from metadata: partition=" + partitionName + ", #files=" + statuses.length);
+ return statuses;
+ }
+
+ /**
+ * Retrieve the merged {@code HoodieRecord} mapped to the given key.
+ *
+ * @param key The key of the record
+ */
+ private Option<HoodieRecord<HoodieMetadataPayload>> getMergedRecordByKey(String key) throws IOException {
+ openTimelineScanner();
+
+ Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
+
+ // Retrieve record from unsynced timeline instants
+ Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
+ if (timelineHoodieRecord.isPresent()) {
+ if (metadataHoodieRecord.isPresent()) {
+ HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
+ metadataHoodieRecord = Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
+ } else {
+ metadataHoodieRecord = timelineHoodieRecord;
+ }
+ }
+
+ return metadataHoodieRecord;
+ }
+
+ protected abstract Option<HoodieRecord<HoodieMetadataPayload>> getRecordByKeyFromMetadata(String key) throws IOException;
+
+ private void openTimelineScanner() throws IOException {
+ if (timelineRecordScanner != null) {
+ // Already opened
+ return;
+ }
+
+ HoodieTableMetaClient datasetMetaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<HoodieInstant> unsyncedInstants = findInstantsToSync(datasetMetaClient);
+ Schema schema = HoodieAvroUtils.addMetadataFields(HoodieMetadataRecord.getClassSchema());
+ timelineRecordScanner =
+ new HoodieMetadataMergedInstantRecordScanner(datasetMetaClient, unsyncedInstants, schema, MAX_MEMORY_SIZE_IN_BYTES, spillableMapDirectory, null);
Review comment:
Filter would be beneficial if we are looking for a specific key. However case often when reading from metadata table is reading multiple keys so is beneficial to have everything in this case
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.metadata;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+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.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CleanerUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.metadata.HoodieTableMetadata.NON_PARTITIONED_NAME;
+
+/**
+ * A utility to convert timeline information to metadata table records.
+ */
+public class HoodieTableMetadataUtil {
+
+ private static final Logger LOG = LogManager.getLogger(HoodieTableMetadataUtil.class);
+
+ /**
+ * Converts a timeline instant to metadata table records.
+ *
+ * @param datasetMetaClient The meta client associated with the timeline instant
+ * @param instant to fetch and convert to metadata table records
+ * @return a list of metadata table records
+ * @throws IOException
+ */
+ public static Option<List<HoodieRecord>> convertInstantToMetaRecords(HoodieTableMetaClient datasetMetaClient, HoodieInstant instant) throws IOException {
Review comment:
Correct other then the callout I mentioned
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r547573663
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMergedInstantRecordScanner.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.DefaultSizeEstimator;
+import org.apache.hudi.common.util.HoodieRecordSizeEstimator;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ExternalSpillableMap;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Provides functionality to convert timeline instants to table metadata records and then merge by key. Specify
+ * a filter to limit keys that are merged and stored in memory.
+ */
+public class HoodieMetadataMergedInstantRecordScanner {
+
+ private static final Logger LOG = LogManager.getLogger(HoodieMetadataMergedInstantRecordScanner.class);
+
+ HoodieTableMetaClient metaClient;
+ private List<HoodieInstant> instants;
+ private Set<String> mergeKeyFilter;
+ protected final ExternalSpillableMap<String, HoodieRecord<? extends HoodieRecordPayload>> records;
+
+ public HoodieMetadataMergedInstantRecordScanner(HoodieTableMetaClient metaClient, List<HoodieInstant> instants,
+ Schema readerSchema, Long maxMemorySizeInBytes,
+ String spillableMapBasePath, Set<String> mergeKeyFilter) throws IOException {
+ this.metaClient = metaClient;
+ this.instants = instants;
+ this.mergeKeyFilter = mergeKeyFilter != null ? mergeKeyFilter : Collections.emptySet();
+ this.records = new ExternalSpillableMap<>(maxMemorySizeInBytes, spillableMapBasePath, new DefaultSizeEstimator(),
+ new HoodieRecordSizeEstimator(readerSchema));
+
+ scan();
+ }
+
+ private void scan() {
+ for (HoodieInstant instant : instants) {
+ try {
+ processInstant(instant);
+ } catch (Exception e) {
+ LOG.error(String.format("Got exception when processing timeline instant %s", instant.getTimestamp()), e);
+ throw new HoodieException(String.format("Got exception when processing timeline instant %s", instant.getTimestamp()), e);
+ }
+ }
+ }
+
+ /**
+ * Converts an instant to metadata table records and processes each record.
+ *
+ * @param instant
+ * @throws IOException
+ */
+ private void processInstant(HoodieInstant instant) throws IOException {
Review comment:
Updated
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -372,19 +191,6 @@ protected void closeReaders() {
logRecordScanner = null;
}
Review comment:
Thanks for the callout. I added this to `closeReaders()`
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/AbstractHoodieTableMetadata.java
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.avro.model.HoodieMetadataRecord;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.model.HoodiePartitionMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Interface that supports querying various pieces of metadata about a hudi table.
+ */
+public abstract class AbstractHoodieTableMetadata implements HoodieTableMetadata {
+
+ private static final Logger LOG = LogManager.getLogger(AbstractHoodieTableMetadata.class);
+
+ static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
+ static final int BUFFER_SIZE = 10 * 1024 * 1024;
+
+ protected final SerializableConfiguration hadoopConf;
+ protected final Option<HoodieMetadataMetrics> metrics;
+ protected final String datasetBasePath;
+
+ // Directory used for Spillable Map when merging records
+ final String spillableMapDirectory;
+
+ protected boolean enabled;
+ private final boolean validateLookups;
+ private final boolean assumeDatePartitioning;
+
+ private transient HoodieMetadataMergedInstantRecordScanner timelineRecordScanner;
+
+ protected AbstractHoodieTableMetadata(Configuration hadoopConf, String datasetBasePath, String spillableMapDirectory,
+ boolean enabled, boolean validateLookups, boolean enableMetrics,
+ boolean assumeDatePartitioning) {
+ this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this.datasetBasePath = datasetBasePath;
+ this.spillableMapDirectory = spillableMapDirectory;
+
+ this.enabled = enabled;
+ this.validateLookups = validateLookups;
+ this.assumeDatePartitioning = assumeDatePartitioning;
+
+ if (enableMetrics) {
+ this.metrics = Option.of(new HoodieMetadataMetrics(Registry.getRegistry("HoodieMetadata")));
+ } else {
+ this.metrics = Option.empty();
+ }
+ }
+
+ public static AbstractHoodieTableMetadata create(Configuration conf, String datasetBasePath, String spillableMapPath, boolean useFileListingFromMetadata,
+ boolean verifyListings, boolean enableMetrics, boolean shouldAssumeDatePartitioning) {
+ return new HoodieBackedTableMetadata(conf, datasetBasePath, spillableMapPath, useFileListingFromMetadata, verifyListings,
+ enableMetrics, shouldAssumeDatePartitioning);
+ }
+
+
+ /**
+ * Return the list of files in a partition.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ * @param partitionPath The absolute path of the partition to list
+ */
+
+ public FileStatus[] getAllFilesInPartition(Path partitionPath) throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllFilesInPartition(partitionPath);
+ } catch (Exception e) {
+ LOG.error("Failed to retrive files in partition " + partitionPath + " from metadata", e);
+ }
+ }
+ return FSUtils.getFs(partitionPath.toString(), hadoopConf.get()).listStatus(partitionPath);
+ }
+
+ /**
+ * Return the list of partitions in the dataset.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ */
+ public List<String> getAllPartitionPaths() throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllPartitionPaths();
+ } catch (Exception e) {
+ LOG.error("Failed to retrieve list of partition from metadata", e);
+ }
+ }
+
+ FileSystem fs = FSUtils.getFs(datasetBasePath, hadoopConf.get());
+ return FSUtils.getAllPartitionPaths(fs, datasetBasePath, assumeDatePartitioning);
+ }
+
+ /**
+ * Returns a list of all partitions.
+ */
+ /**
+ * Returns a list of all partitions.
+ */
+ protected List<String> fetchAllPartitionPaths() throws IOException {
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(RECORDKEY_PARTITION_LIST);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_PARTITIONS_STR, timer.endTimer()));
+
+ List<String> partitions = Collections.emptyList();
+ if (hoodieRecord.isPresent()) {
+ if (!hoodieRecord.get().getData().getDeletions().isEmpty()) {
+ throw new HoodieMetadataException("Metadata partition list record is inconsistent: "
+ + hoodieRecord.get().getData());
+ }
+
+ partitions = hoodieRecord.get().getData().getFilenames();
+ // Partition-less tables have a single empty partition
+ if (partitions.contains(NON_PARTITIONED_NAME)) {
+ partitions.remove(NON_PARTITIONED_NAME);
+ partitions.add("");
+ }
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<String> actualPartitions = FSUtils.getAllPartitionPaths(metaClient.getFs(), datasetBasePath, false);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_PARTITIONS_STR, timer.endTimer()));
+
+ Collections.sort(actualPartitions);
+ Collections.sort(partitions);
+ if (!actualPartitions.equals(partitions)) {
+ LOG.error("Validation of metadata partition list failed. Lists do not match.");
+ LOG.error("Partitions from metadata: " + Arrays.toString(partitions.toArray()));
+ LOG.error("Partitions from file system: " + Arrays.toString(actualPartitions.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ partitions = actualPartitions;
+ }
+
+ LOG.info("Listed partitions from metadata: #partitions=" + partitions.size());
+ return partitions;
+ }
+
+ /**
+ * Return all the files from the partition.
+ *
+ * @param partitionPath The absolute path of the partition
+ */
+ private FileStatus[] fetchAllFilesInPartition(Path partitionPath) throws IOException {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(datasetBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ partitionName = NON_PARTITIONED_NAME;
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(partitionName);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+
+ FileStatus[] statuses = {};
+ if (hoodieRecord.isPresent()) {
+ statuses = hoodieRecord.get().getData().getFileStatuses(partitionPath);
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+
+ // Ignore partition metadata file
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ FileStatus[] directStatuses = metaClient.getFs().listStatus(partitionPath,
+ p -> !p.getName().equals(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE));
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_FILES_STR, timer.endTimer()));
+
+ List<String> directFilenames = Arrays.stream(directStatuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ List<String> metadataFilenames = Arrays.stream(statuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ if (!metadataFilenames.equals(directFilenames)) {
+ LOG.error("Validation of metadata file listing for partition " + partitionName + " failed.");
+ LOG.error("File list from metadata: " + Arrays.toString(metadataFilenames.toArray()));
+ LOG.error("File list from direct listing: " + Arrays.toString(directFilenames.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ statuses = directStatuses;
+ }
+
+ LOG.info("Listed file in partition from metadata: partition=" + partitionName + ", #files=" + statuses.length);
+ return statuses;
+ }
+
+ /**
+ * Retrieve the merged {@code HoodieRecord} mapped to the given key.
+ *
+ * @param key The key of the record
+ */
+ private Option<HoodieRecord<HoodieMetadataPayload>> getMergedRecordByKey(String key) throws IOException {
+ openTimelineScanner();
+
+ Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
+
+ // Retrieve record from unsynced timeline instants
+ Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
+ if (timelineHoodieRecord.isPresent()) {
+ if (metadataHoodieRecord.isPresent()) {
+ HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
+ metadataHoodieRecord = Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
+ } else {
+ metadataHoodieRecord = timelineHoodieRecord;
Review comment:
Sorry my comment was unclear. I created separate `mergedRecord` variable so I am not overriding `metadataHoodieRecord`
##########
File path: hudi-client/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
##########
@@ -266,7 +256,7 @@ private void initialize(JavaSparkContext jsc, HoodieTableMetaClient datasetMetaC
}
private void initTableMetadata() {
- this.metadata = new HoodieBackedTableMetadata(hadoopConf.get(), datasetWriteConfig.getBasePath(), datasetWriteConfig.getSpillableMapBasePath(),
+ this.metadata = (HoodieBackedTableMetadata) AbstractHoodieTableMetadata.create(hadoopConf.get(), datasetWriteConfig.getBasePath(), datasetWriteConfig.getSpillableMapBasePath(),
Review comment:
Along with Prashant's comment we can just create `HoodieBackedTableMetadata` since `HoodieBackedTableMetadaWriter` should only be associated with this implementation
##########
File path: hudi-client/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
##########
@@ -266,7 +256,7 @@ private void initialize(JavaSparkContext jsc, HoodieTableMetaClient datasetMetaC
}
private void initTableMetadata() {
- this.metadata = new HoodieBackedTableMetadata(hadoopConf.get(), datasetWriteConfig.getBasePath(), datasetWriteConfig.getSpillableMapBasePath(),
Review comment:
Agreed. Updated this
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r546309417
##########
File path: hudi-client/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
##########
@@ -266,7 +256,7 @@ private void initialize(JavaSparkContext jsc, HoodieTableMetaClient datasetMetaC
}
private void initTableMetadata() {
- this.metadata = new HoodieBackedTableMetadata(hadoopConf.get(), datasetWriteConfig.getBasePath(), datasetWriteConfig.getSpillableMapBasePath(),
+ this.metadata = (HoodieBackedTableMetadata) AbstractHoodieTableMetadata.create(hadoopConf.get(), datasetWriteConfig.getBasePath(), datasetWriteConfig.getSpillableMapBasePath(),
Review comment:
I think we can keep the factory method in the interface?
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -66,24 +58,13 @@
* If the metadata table does not exist, RPC calls are used to retrieve file listings from the file system.
* No updates are applied to the table and it is not synced.
*/
-public class HoodieBackedTableMetadata implements HoodieTableMetadata {
+public class HoodieBackedTableMetadata extends AbstractHoodieTableMetadata {
Review comment:
Given the only subclass of `AbstractHoodieTableMetadata` is this class and the abstract class already assumes the existence of the metadata table, I am not sure if the splitting is needed per se.
I was under the impression that we would like to implement the timeline merging as another subclass. If we stick with the new scanner approach (which I actually have grown to like actually), we no longer need the abstract class right?
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.metadata;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+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.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CleanerUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.metadata.HoodieTableMetadata.NON_PARTITIONED_NAME;
+
+/**
+ * A utility to convert timeline information to metadata table records.
+ */
+public class HoodieTableMetadataUtil {
+
+ private static final Logger LOG = LogManager.getLogger(HoodieTableMetadataUtil.class);
+
+ /**
+ * Converts a timeline instant to metadata table records.
+ *
+ * @param datasetMetaClient The meta client associated with the timeline instant
+ * @param instant to fetch and convert to metadata table records
+ * @return a list of metadata table records
+ * @throws IOException
+ */
+ public static Option<List<HoodieRecord>> convertInstantToMetaRecords(HoodieTableMetaClient datasetMetaClient, HoodieInstant instant) throws IOException {
Review comment:
assume most of this code is just from the writer class verbatim
##########
File path: hudi-client/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
##########
@@ -569,114 +471,7 @@ public void update(HoodieRollbackMetadata rollbackMetadata, String instantTime)
return;
}
- Map<String, Map<String, Long>> partitionToAppendedFiles = new HashMap<>();
- Map<String, List<String>> partitionToDeletedFiles = new HashMap<>();
- processRollbackMetadata(rollbackMetadata, partitionToDeletedFiles, partitionToAppendedFiles);
- commitRollback(jsc, partitionToDeletedFiles, partitionToAppendedFiles, instantTime, "Rollback");
- }
-
- /**
- * Extracts information about the deleted and append files from the {@code HoodieRollbackMetadata}.
- *
- * During a rollback files may be deleted (COW, MOR) or rollback blocks be appended (MOR only) to files. This
- * function will extract this change file for each partition.
- *
- * @param rollbackMetadata {@code HoodieRollbackMetadata}
- * @param partitionToDeletedFiles The {@code Map} to fill with files deleted per partition.
- * @param partitionToAppendedFiles The {@code Map} to fill with files appended per partition and their sizes.
- */
- private void processRollbackMetadata(HoodieRollbackMetadata rollbackMetadata,
- Map<String, List<String>> partitionToDeletedFiles,
- Map<String, Map<String, Long>> partitionToAppendedFiles) {
- rollbackMetadata.getPartitionMetadata().values().forEach(pm -> {
- final String partition = pm.getPartitionPath();
-
- if (!pm.getSuccessDeleteFiles().isEmpty()) {
- if (!partitionToDeletedFiles.containsKey(partition)) {
- partitionToDeletedFiles.put(partition, new ArrayList<>());
- }
-
- // Extract deleted file name from the absolute paths saved in getSuccessDeleteFiles()
- List<String> deletedFiles = pm.getSuccessDeleteFiles().stream().map(p -> new Path(p).getName())
- .collect(Collectors.toList());
- partitionToDeletedFiles.get(partition).addAll(deletedFiles);
- }
-
- if (!pm.getAppendFiles().isEmpty()) {
- if (!partitionToAppendedFiles.containsKey(partition)) {
- partitionToAppendedFiles.put(partition, new HashMap<>());
- }
-
- // Extract appended file name from the absolute paths saved in getAppendFiles()
- pm.getAppendFiles().forEach((path, size) -> {
- partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> {
- return size + oldSize;
- });
- });
- }
- });
- }
-
- /**
- * Create file delete records and commit.
- *
- * @param partitionToDeletedFiles {@code Map} of partitions and the deleted files
- * @param instantTime Timestamp at which the deletes took place
- * @param operation Type of the operation which caused the files to be deleted
- */
- private void commitRollback(JavaSparkContext jsc, Map<String, List<String>> partitionToDeletedFiles,
- Map<String, Map<String, Long>> partitionToAppendedFiles, String instantTime,
- String operation) {
- List<HoodieRecord> records = new LinkedList<>();
- int[] fileChangeCount = {0, 0}; // deletes, appends
-
- partitionToDeletedFiles.forEach((partition, deletedFiles) -> {
- // Rollbacks deletes instants from timeline. The instant being rolled-back may not have been synced to the
Review comment:
left a comment around this scenario. there is one valid case here. good call out in the summary
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -117,172 +93,15 @@ public HoodieBackedTableMetadata(Configuration conf, String datasetBasePath, Str
} else {
LOG.info("Metadata table is disabled.");
}
-
Review comment:
I assume all of this code, is just verbatim moved up to the base class ?
##########
File path: hudi-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java
##########
@@ -619,32 +618,97 @@ public void testMetadataMetrics() throws Exception {
}
}
+ //@ParameterizedTest
+ //@EnumSource(HoodieTableType.class)
+ public void testMetadataOutOfSync(HoodieTableType tableType) throws Exception {
+ init(HoodieTableType.COPY_ON_WRITE);
+
+ HoodieWriteClient unsyncedClient = new HoodieWriteClient<>(jsc, getWriteConfig(true, true));
+
+ // Enable metadata so table is initialized
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, true))) {
+ // Perform Bulk Insert
+ String newCommitTime = "001";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.bulkInsert(jsc.parallelize(records, 1), newCommitTime).collect();
+ }
+
+ // Perform commit operations with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ // Perform Insert
+ String newCommitTime = "002";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.insert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Perform Upsert
+ newCommitTime = "003";
+ client.startCommitWithTime(newCommitTime);
+ records = dataGen.generateUniqueUpdates(newCommitTime, 20);
+ client.upsert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Compaction
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ newCommitTime = "004";
+ client.scheduleCompactionAtInstant(newCommitTime, Option.empty());
+ client.compact(newCommitTime);
+ }
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
+ validateMetadata(unsyncedClient);
+
+ // Perform clean operation with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ // One more commit needed to trigger clean so upsert and compact
+ String newCommitTime = "005";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateUpdates(newCommitTime, 20);
+ client.upsert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ newCommitTime = "006";
+ client.scheduleCompactionAtInstant(newCommitTime, Option.empty());
+ client.compact(newCommitTime);
+ }
+
+ // Clean
+ newCommitTime = "007";
+ client.clean(newCommitTime);
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
+ validateMetadata(unsyncedClient);
+
+ // Perform restore with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ client.restoreToInstant("004");
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
+ validateMetadata(unsyncedClient);
+ }
+
/**
* Validate the metadata tables contents to ensure it matches what is on the file system.
*
* @throws IOException
*/
private void validateMetadata(HoodieWriteClient client) throws IOException {
HoodieWriteConfig config = client.getConfig();
- HoodieBackedTableMetadataWriter metadataWriter = metadataWriter(client);
- assertNotNull(metadataWriter, "MetadataWriter should have been initialized");
+
+ HoodieBackedTableMetadata metadataReader = metadata(client);
Review comment:
just `tableMetadata`?
##########
File path: hudi-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java
##########
@@ -619,32 +618,97 @@ public void testMetadataMetrics() throws Exception {
}
}
+ //@ParameterizedTest
+ //@EnumSource(HoodieTableType.class)
+ public void testMetadataOutOfSync(HoodieTableType tableType) throws Exception {
+ init(HoodieTableType.COPY_ON_WRITE);
+
+ HoodieWriteClient unsyncedClient = new HoodieWriteClient<>(jsc, getWriteConfig(true, true));
+
+ // Enable metadata so table is initialized
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, true))) {
+ // Perform Bulk Insert
+ String newCommitTime = "001";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.bulkInsert(jsc.parallelize(records, 1), newCommitTime).collect();
+ }
+
+ // Perform commit operations with metadata disabled
+ try (HoodieWriteClient client = new HoodieWriteClient<>(jsc, getWriteConfig(true, false))) {
+ // Perform Insert
+ String newCommitTime = "002";
+ client.startCommitWithTime(newCommitTime);
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 20);
+ client.insert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Perform Upsert
+ newCommitTime = "003";
+ client.startCommitWithTime(newCommitTime);
+ records = dataGen.generateUniqueUpdates(newCommitTime, 20);
+ client.upsert(jsc.parallelize(records, 1), newCommitTime).collect();
+
+ // Compaction
+ if (metaClient.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ newCommitTime = "004";
+ client.scheduleCompactionAtInstant(newCommitTime, Option.empty());
+ client.compact(newCommitTime);
+ }
+ }
+
+ assertFalse(metadata(unsyncedClient).isInSync());
Review comment:
won't metadata be synced in `postCommit()` of the last upsert above. wondering how this is actually not in sync
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/AbstractHoodieTableMetadata.java
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.avro.model.HoodieMetadataRecord;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.model.HoodiePartitionMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Interface that supports querying various pieces of metadata about a hudi table.
+ */
+public abstract class AbstractHoodieTableMetadata implements HoodieTableMetadata {
+
+ private static final Logger LOG = LogManager.getLogger(AbstractHoodieTableMetadata.class);
+
+ static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
+ static final int BUFFER_SIZE = 10 * 1024 * 1024;
+
+ protected final SerializableConfiguration hadoopConf;
+ protected final Option<HoodieMetadataMetrics> metrics;
+ protected final String datasetBasePath;
+
+ // Directory used for Spillable Map when merging records
+ final String spillableMapDirectory;
+
+ protected boolean enabled;
+ private final boolean validateLookups;
+ private final boolean assumeDatePartitioning;
+
+ private transient HoodieMetadataMergedInstantRecordScanner timelineRecordScanner;
+
+ protected AbstractHoodieTableMetadata(Configuration hadoopConf, String datasetBasePath, String spillableMapDirectory,
+ boolean enabled, boolean validateLookups, boolean enableMetrics,
+ boolean assumeDatePartitioning) {
+ this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this.datasetBasePath = datasetBasePath;
+ this.spillableMapDirectory = spillableMapDirectory;
+
+ this.enabled = enabled;
+ this.validateLookups = validateLookups;
+ this.assumeDatePartitioning = assumeDatePartitioning;
+
+ if (enableMetrics) {
+ this.metrics = Option.of(new HoodieMetadataMetrics(Registry.getRegistry("HoodieMetadata")));
+ } else {
+ this.metrics = Option.empty();
+ }
+ }
+
+ public static AbstractHoodieTableMetadata create(Configuration conf, String datasetBasePath, String spillableMapPath, boolean useFileListingFromMetadata,
+ boolean verifyListings, boolean enableMetrics, boolean shouldAssumeDatePartitioning) {
+ return new HoodieBackedTableMetadata(conf, datasetBasePath, spillableMapPath, useFileListingFromMetadata, verifyListings,
+ enableMetrics, shouldAssumeDatePartitioning);
+ }
+
+
+ /**
+ * Return the list of files in a partition.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ * @param partitionPath The absolute path of the partition to list
+ */
+
+ public FileStatus[] getAllFilesInPartition(Path partitionPath) throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllFilesInPartition(partitionPath);
+ } catch (Exception e) {
+ LOG.error("Failed to retrive files in partition " + partitionPath + " from metadata", e);
+ }
+ }
+ return FSUtils.getFs(partitionPath.toString(), hadoopConf.get()).listStatus(partitionPath);
+ }
+
+ /**
+ * Return the list of partitions in the dataset.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ */
+ public List<String> getAllPartitionPaths() throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllPartitionPaths();
+ } catch (Exception e) {
+ LOG.error("Failed to retrieve list of partition from metadata", e);
+ }
+ }
+
+ FileSystem fs = FSUtils.getFs(datasetBasePath, hadoopConf.get());
+ return FSUtils.getAllPartitionPaths(fs, datasetBasePath, assumeDatePartitioning);
+ }
+
+ /**
+ * Returns a list of all partitions.
+ */
+ /**
+ * Returns a list of all partitions.
+ */
+ protected List<String> fetchAllPartitionPaths() throws IOException {
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(RECORDKEY_PARTITION_LIST);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_PARTITIONS_STR, timer.endTimer()));
+
+ List<String> partitions = Collections.emptyList();
+ if (hoodieRecord.isPresent()) {
+ if (!hoodieRecord.get().getData().getDeletions().isEmpty()) {
+ throw new HoodieMetadataException("Metadata partition list record is inconsistent: "
+ + hoodieRecord.get().getData());
+ }
+
+ partitions = hoodieRecord.get().getData().getFilenames();
+ // Partition-less tables have a single empty partition
+ if (partitions.contains(NON_PARTITIONED_NAME)) {
+ partitions.remove(NON_PARTITIONED_NAME);
+ partitions.add("");
+ }
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<String> actualPartitions = FSUtils.getAllPartitionPaths(metaClient.getFs(), datasetBasePath, false);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_PARTITIONS_STR, timer.endTimer()));
+
+ Collections.sort(actualPartitions);
+ Collections.sort(partitions);
+ if (!actualPartitions.equals(partitions)) {
+ LOG.error("Validation of metadata partition list failed. Lists do not match.");
+ LOG.error("Partitions from metadata: " + Arrays.toString(partitions.toArray()));
+ LOG.error("Partitions from file system: " + Arrays.toString(actualPartitions.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ partitions = actualPartitions;
+ }
+
+ LOG.info("Listed partitions from metadata: #partitions=" + partitions.size());
+ return partitions;
+ }
+
+ /**
+ * Return all the files from the partition.
+ *
+ * @param partitionPath The absolute path of the partition
+ */
+ private FileStatus[] fetchAllFilesInPartition(Path partitionPath) throws IOException {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(datasetBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ partitionName = NON_PARTITIONED_NAME;
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(partitionName);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+
+ FileStatus[] statuses = {};
+ if (hoodieRecord.isPresent()) {
+ statuses = hoodieRecord.get().getData().getFileStatuses(partitionPath);
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+
+ // Ignore partition metadata file
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ FileStatus[] directStatuses = metaClient.getFs().listStatus(partitionPath,
+ p -> !p.getName().equals(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE));
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_FILES_STR, timer.endTimer()));
+
+ List<String> directFilenames = Arrays.stream(directStatuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ List<String> metadataFilenames = Arrays.stream(statuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ if (!metadataFilenames.equals(directFilenames)) {
+ LOG.error("Validation of metadata file listing for partition " + partitionName + " failed.");
+ LOG.error("File list from metadata: " + Arrays.toString(metadataFilenames.toArray()));
+ LOG.error("File list from direct listing: " + Arrays.toString(directFilenames.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ statuses = directStatuses;
+ }
+
+ LOG.info("Listed file in partition from metadata: partition=" + partitionName + ", #files=" + statuses.length);
+ return statuses;
+ }
+
+ /**
+ * Retrieve the merged {@code HoodieRecord} mapped to the given key.
+ *
+ * @param key The key of the record
+ */
+ private Option<HoodieRecord<HoodieMetadataPayload>> getMergedRecordByKey(String key) throws IOException {
+ openTimelineScanner();
+
+ Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
+
+ // Retrieve record from unsynced timeline instants
+ Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
+ if (timelineHoodieRecord.isPresent()) {
+ if (metadataHoodieRecord.isPresent()) {
+ HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
+ metadataHoodieRecord = Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
+ } else {
+ metadataHoodieRecord = timelineHoodieRecord;
Review comment:
just return out of here, instead of reassigning to another important variable. may be easier to read.
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/AbstractHoodieTableMetadata.java
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.metadata;
+
+import org.apache.avro.Schema;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.avro.model.HoodieMetadataRecord;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.model.HoodiePartitionMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordPayload;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Interface that supports querying various pieces of metadata about a hudi table.
+ */
+public abstract class AbstractHoodieTableMetadata implements HoodieTableMetadata {
+
+ private static final Logger LOG = LogManager.getLogger(AbstractHoodieTableMetadata.class);
+
+ static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
+ static final int BUFFER_SIZE = 10 * 1024 * 1024;
+
+ protected final SerializableConfiguration hadoopConf;
+ protected final Option<HoodieMetadataMetrics> metrics;
+ protected final String datasetBasePath;
+
+ // Directory used for Spillable Map when merging records
+ final String spillableMapDirectory;
+
+ protected boolean enabled;
+ private final boolean validateLookups;
+ private final boolean assumeDatePartitioning;
+
+ private transient HoodieMetadataMergedInstantRecordScanner timelineRecordScanner;
+
+ protected AbstractHoodieTableMetadata(Configuration hadoopConf, String datasetBasePath, String spillableMapDirectory,
+ boolean enabled, boolean validateLookups, boolean enableMetrics,
+ boolean assumeDatePartitioning) {
+ this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this.datasetBasePath = datasetBasePath;
+ this.spillableMapDirectory = spillableMapDirectory;
+
+ this.enabled = enabled;
+ this.validateLookups = validateLookups;
+ this.assumeDatePartitioning = assumeDatePartitioning;
+
+ if (enableMetrics) {
+ this.metrics = Option.of(new HoodieMetadataMetrics(Registry.getRegistry("HoodieMetadata")));
+ } else {
+ this.metrics = Option.empty();
+ }
+ }
+
+ public static AbstractHoodieTableMetadata create(Configuration conf, String datasetBasePath, String spillableMapPath, boolean useFileListingFromMetadata,
+ boolean verifyListings, boolean enableMetrics, boolean shouldAssumeDatePartitioning) {
+ return new HoodieBackedTableMetadata(conf, datasetBasePath, spillableMapPath, useFileListingFromMetadata, verifyListings,
+ enableMetrics, shouldAssumeDatePartitioning);
+ }
+
+
+ /**
+ * Return the list of files in a partition.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ * @param partitionPath The absolute path of the partition to list
+ */
+
+ public FileStatus[] getAllFilesInPartition(Path partitionPath) throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllFilesInPartition(partitionPath);
+ } catch (Exception e) {
+ LOG.error("Failed to retrive files in partition " + partitionPath + " from metadata", e);
+ }
+ }
+ return FSUtils.getFs(partitionPath.toString(), hadoopConf.get()).listStatus(partitionPath);
+ }
+
+ /**
+ * Return the list of partitions in the dataset.
+ *
+ * If the Metadata Table is enabled, the listing is retrieved from the stored metadata. Otherwise, the list of
+ * partitions is retrieved directly from the underlying {@code FileSystem}.
+ *
+ * On any errors retrieving the listing from the metadata, defaults to using the file system listings.
+ *
+ */
+ public List<String> getAllPartitionPaths() throws IOException {
+ if (enabled) {
+ try {
+ return fetchAllPartitionPaths();
+ } catch (Exception e) {
+ LOG.error("Failed to retrieve list of partition from metadata", e);
+ }
+ }
+
+ FileSystem fs = FSUtils.getFs(datasetBasePath, hadoopConf.get());
+ return FSUtils.getAllPartitionPaths(fs, datasetBasePath, assumeDatePartitioning);
+ }
+
+ /**
+ * Returns a list of all partitions.
+ */
+ /**
+ * Returns a list of all partitions.
+ */
+ protected List<String> fetchAllPartitionPaths() throws IOException {
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(RECORDKEY_PARTITION_LIST);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_PARTITIONS_STR, timer.endTimer()));
+
+ List<String> partitions = Collections.emptyList();
+ if (hoodieRecord.isPresent()) {
+ if (!hoodieRecord.get().getData().getDeletions().isEmpty()) {
+ throw new HoodieMetadataException("Metadata partition list record is inconsistent: "
+ + hoodieRecord.get().getData());
+ }
+
+ partitions = hoodieRecord.get().getData().getFilenames();
+ // Partition-less tables have a single empty partition
+ if (partitions.contains(NON_PARTITIONED_NAME)) {
+ partitions.remove(NON_PARTITIONED_NAME);
+ partitions.add("");
+ }
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<String> actualPartitions = FSUtils.getAllPartitionPaths(metaClient.getFs(), datasetBasePath, false);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_PARTITIONS_STR, timer.endTimer()));
+
+ Collections.sort(actualPartitions);
+ Collections.sort(partitions);
+ if (!actualPartitions.equals(partitions)) {
+ LOG.error("Validation of metadata partition list failed. Lists do not match.");
+ LOG.error("Partitions from metadata: " + Arrays.toString(partitions.toArray()));
+ LOG.error("Partitions from file system: " + Arrays.toString(actualPartitions.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ partitions = actualPartitions;
+ }
+
+ LOG.info("Listed partitions from metadata: #partitions=" + partitions.size());
+ return partitions;
+ }
+
+ /**
+ * Return all the files from the partition.
+ *
+ * @param partitionPath The absolute path of the partition
+ */
+ private FileStatus[] fetchAllFilesInPartition(Path partitionPath) throws IOException {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(datasetBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ partitionName = NON_PARTITIONED_NAME;
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ Option<HoodieRecord<HoodieMetadataPayload>> hoodieRecord = getMergedRecordByKey(partitionName);
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+
+ FileStatus[] statuses = {};
+ if (hoodieRecord.isPresent()) {
+ statuses = hoodieRecord.get().getData().getFileStatuses(partitionPath);
+ }
+
+ if (validateLookups) {
+ // Validate the Metadata Table data by listing the partitions from the file system
+ timer.startTimer();
+
+ // Ignore partition metadata file
+ HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ FileStatus[] directStatuses = metaClient.getFs().listStatus(partitionPath,
+ p -> !p.getName().equals(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE));
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_FILES_STR, timer.endTimer()));
+
+ List<String> directFilenames = Arrays.stream(directStatuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ List<String> metadataFilenames = Arrays.stream(statuses)
+ .map(s -> s.getPath().getName()).sorted()
+ .collect(Collectors.toList());
+
+ if (!metadataFilenames.equals(directFilenames)) {
+ LOG.error("Validation of metadata file listing for partition " + partitionName + " failed.");
+ LOG.error("File list from metadata: " + Arrays.toString(metadataFilenames.toArray()));
+ LOG.error("File list from direct listing: " + Arrays.toString(directFilenames.toArray()));
+
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.VALIDATE_ERRORS_STR, 0));
+ }
+
+ // Return the direct listing as it should be correct
+ statuses = directStatuses;
+ }
+
+ LOG.info("Listed file in partition from metadata: partition=" + partitionName + ", #files=" + statuses.length);
+ return statuses;
+ }
+
+ /**
+ * Retrieve the merged {@code HoodieRecord} mapped to the given key.
+ *
+ * @param key The key of the record
+ */
+ private Option<HoodieRecord<HoodieMetadataPayload>> getMergedRecordByKey(String key) throws IOException {
+ openTimelineScanner();
+
+ Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
+
+ // Retrieve record from unsynced timeline instants
+ Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
+ if (timelineHoodieRecord.isPresent()) {
+ if (metadataHoodieRecord.isPresent()) {
+ HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
+ metadataHoodieRecord = Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
+ } else {
+ metadataHoodieRecord = timelineHoodieRecord;
+ }
+ }
+
+ return metadataHoodieRecord;
+ }
+
+ protected abstract Option<HoodieRecord<HoodieMetadataPayload>> getRecordByKeyFromMetadata(String key) throws IOException;
+
+ private void openTimelineScanner() throws IOException {
+ if (timelineRecordScanner != null) {
+ // Already opened
+ return;
+ }
+
+ HoodieTableMetaClient datasetMetaClient = new HoodieTableMetaClient(hadoopConf.get(), datasetBasePath);
+ List<HoodieInstant> unsyncedInstants = findInstantsToSync(datasetMetaClient);
+ Schema schema = HoodieAvroUtils.addMetadataFields(HoodieMetadataRecord.getClassSchema());
+ timelineRecordScanner =
+ new HoodieMetadataMergedInstantRecordScanner(datasetMetaClient, unsyncedInstants, schema, MAX_MEMORY_SIZE_IN_BYTES, spillableMapDirectory, null);
Review comment:
should we pass in the filter? guess it does not matter, since we ll read the instants in full anyway
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-750456129
> ```
> data timeline : C1, rollback C12, rollback C11, ..... rollback C2
> metadata timeline: C3, ..., C8
> ```
>
> It's now hard to determine whether or not `rollback C2` should be applied. (we could compare the latest instant on metadata timeline C8 and assume its applied?)
Regarding this scenario, my thoughts would be to check rollback metadata only against unsynced instants, so only after C8. Is this potentially surfacing a more general issue though. Say metadata timeline has been synced up to C2, but data timeline archival has happened to C8. It seems next time we try to find unsynced instants we won't consider anything between C2...C8
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-751585060
Apologies for the delay in pushing changes, got caught up in the holidays :). Updated to address feedback and to resolve our discussions to check whether the commit being rolled back is after last sync point on metadata timeline
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r545511401
##########
File path: hudi-client/src/main/java/org/apache/hudi/table/HoodieTable.java
##########
@@ -635,9 +635,9 @@ public boolean requireSortedRecords() {
return getBaseFileFormat() == HoodieFileFormat.HFILE;
}
- public HoodieTableMetadata metadata() {
+ public AbstractHoodieTableMetadata metadata() {
Review comment:
I maintained this interface in revision after rebasing and then built `AbstractHoodieTableMetadata` on top of this.
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataTimelineUtil.java
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.metadata;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieRecord;
+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.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.util.CleanerUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.metadata.AbstractHoodieTableMetadata.NON_PARTITIONED_NAME;
+
+/**
+ * A utility to convert timeline information to metadata table records.
+ */
+public class HoodieTableMetadataTimelineUtil {
Review comment:
Updated this
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] vinothchandar commented on a change in pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
vinothchandar commented on a change in pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#discussion_r547152380
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -66,24 +58,13 @@
* If the metadata table does not exist, RPC calls are used to retrieve file listings from the file system.
* No updates are applied to the table and it is not synced.
*/
-public class HoodieBackedTableMetadata implements HoodieTableMetadata {
+public class HoodieBackedTableMetadata extends AbstractHoodieTableMetadata {
Review comment:
okay that also makes sense. Personally, I would not have done the split, until a second kind of metadata impl shows up. but udit added the FileSytemBackedMetadata, it does not need the merging per se.. Still we can keep it like this for now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] rmpifer commented on pull request #2342: [RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table
Posted by GitBox <gi...@apache.org>.
rmpifer commented on pull request #2342:
URL: https://github.com/apache/hudi/pull/2342#issuecomment-749217844
> @rmpifer On this point specifically,
>
> > so there would never be the case where a rollback was being written before an instant earlier on the timeline was already synced. Removed this logic because it created circular dependency when implementing timeline merging
>
> Following situation is technically possible. Timeline has commits c1, c2, c3, c4 and we have synced only upto C2. User decides to restore the table to commit time c2, thus generating rollback metadata for c3, c4 which now have no files in the metadata table. I think we should defensively handle this scenario at the payload merge level i.e if there is a delete for a file that does not exist in the base file or prior instants, we should just do a no-op and continue. What I am saying is : I agree with you that the extra logic need not be present, but we should check the payload once to ensure we can handle the scenario above.
>
> Let me know if we are not talking about the same thing :)
If this is the case what happens when c3 and c4 are then finally synced? These files would remain in the metadata table even though rollback should have removed them
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org