This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit b38994a22b179e36462f970391004f2a56c32e88 Author: voonhous <[email protected]> AuthorDate: Tue Jun 16 14:14:09 2026 +0800 refactor: Add Lombok annotations to hudi-common module (part 8) (#18957) * refactor: Add Lombok annotations to hudi-common module (part 8) * refactor: Keep explicit equals/hashCode and toString implementations - Restore getClass() based equals/hashCode in HoodieInstant and TimelineLayoutVersion instead of Lombok's instanceof/canEqual semantics - Restore TimelineLayoutVersion#toString: its value is persisted to hoodie.properties via String.valueOf, Lombok's format would corrupt it (cherry picked from commit b140d249b6cc624131ee0c237aa2c46f6e2c442f) --- .../common/table/timeline/BaseHoodieTimeline.java | 13 ++--- .../hudi/common/table/timeline/HoodieInstant.java | 33 +++--------- .../hudi/common/table/timeline/LSMTimeline.java | 11 ++-- .../table/timeline/MetadataConversionUtils.java | 11 ++-- .../common/table/timeline/TimeGeneratorBase.java | 12 ++--- .../common/table/timeline/TimelineDiffHelper.java | 46 +++++------------ .../hudi/common/table/timeline/TimelineLayout.java | 58 +++------------------- .../hudi/common/table/timeline/TimelineUtils.java | 13 +++-- .../timeline/versioning/TimelineLayoutVersion.java | 8 ++- .../timeline/versioning/v1/ActiveTimelineV1.java | 45 +++++++---------- .../versioning/v1/ArchivedTimelineLoaderV1.java | 10 ++-- .../timeline/versioning/v1/ArchivedTimelineV1.java | 11 ++-- .../versioning/v1/CommitMetadataSerDeV1.java | 5 +- .../versioning/v1/CompletionTimeQueryViewV1.java | 9 ++-- .../timeline/versioning/v2/ActiveTimelineV2.java | 48 ++++++++---------- .../versioning/v2/CompletionTimeQueryViewV2.java | 7 +-- .../table/view/AbstractTableFileSystemView.java | 52 ++++++++----------- .../common/table/view/FileSystemViewManager.java | 27 +++++----- .../table/view/HoodieTableFileSystemView.java | 15 ++---- .../IncrementalTimelineSyncFileSystemView.java | 56 ++++++++++----------- .../table/view/PriorityBasedFileSystemView.java | 25 ++++------ .../view/RemoteHoodieTableFileSystemView.java | 6 +-- .../table/view/RocksDbBasedFileSystemView.java | 50 +++++++++---------- .../view/SpillableMapBasedFileSystemView.java | 8 ++- 24 files changed, 216 insertions(+), 363 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java index f32cce52c451..aaf24bf85fee 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java @@ -27,6 +27,8 @@ import org.apache.hudi.common.util.ValidationUtils; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; +import lombok.Getter; + import java.io.IOException; import java.io.InputStream; import java.io.Serializable; @@ -60,6 +62,7 @@ public abstract class BaseHoodieTimeline implements HoodieTimeline { private static final String HASHING_ALGORITHM = "SHA-256"; + @Getter protected transient HoodieInstantReader instantReader; private List<HoodieInstant> instants; // for efficient #contains queries. @@ -70,6 +73,7 @@ public abstract class BaseHoodieTimeline implements HoodieTimeline { private transient volatile Option<HoodieInstant> firstNonSavepointCommit; // for efficient #isBeforeTimelineStartsByCompletionTime private transient volatile Option<HoodieInstant> firstNonSavepointCommitByCompletionTime; + @Getter private String timelineHash; protected TimelineFactory factory; @@ -495,11 +499,6 @@ public abstract class BaseHoodieTimeline implements HoodieTimeline { return containsInstant(instant) || isBeforeTimelineStarts(instant); } - @Override - public String getTimelineHash() { - return timelineHash; - } - @Override public Stream<HoodieInstant> getInstantsAsStream() { return instants.stream(); @@ -665,10 +664,6 @@ public abstract class BaseHoodieTimeline implements HoodieTimeline { return StringUtils.toHexString(md.digest()); } - public HoodieInstantReader getInstantReader() { - return instantReader; - } - /** * Merges the given instant list into one and keep the sequence. */ diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java index 8d874708cbcb..70935d7e40ef 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java @@ -20,6 +20,10 @@ package org.apache.hudi.common.table.timeline; import org.apache.hudi.common.util.StringUtils; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; + import java.io.Serializable; import java.util.Comparator; import java.util.Objects; @@ -28,6 +32,8 @@ import java.util.Objects; * A Hoodie Instant represents a action done on a hoodie table. All actions start with a inflight instant and then * create a completed instant after done. */ +@AllArgsConstructor +@Getter public class HoodieInstant implements Serializable, Comparable<HoodieInstant> { public static final String FILE_NAME_FORMAT_ERROR = "The provided file name %s does not conform to the required format"; @@ -36,10 +42,12 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> { private final State state; private final String action; + @Getter(AccessLevel.NONE) private final String requestedTime; private final String completionTime; // Marker for older formats, we need the state transition time (pre table version 7) private boolean isLegacy = false; + @Getter(AccessLevel.NONE) private final Comparator<HoodieInstant> comparator; public HoodieInstant(State state, String action, String requestTime, Comparator<HoodieInstant> comparator) { @@ -50,15 +58,6 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> { this(state, action, requestTime, completionTime, false, comparator); } - public HoodieInstant(State state, String action, String requestedTime, String completionTime, boolean isLegacy, Comparator<HoodieInstant> comparator) { - this.state = state; - this.action = action; - this.requestedTime = requestedTime; - this.completionTime = completionTime; - this.isLegacy = isLegacy; - this.comparator = comparator; - } - public boolean isCompleted() { return state == State.COMPLETED; } @@ -71,18 +70,10 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> { return state == State.REQUESTED; } - public String getAction() { - return action; - } - public String requestedTime() { return requestedTime; } - public boolean isLegacy() { - return isLegacy; - } - @Override public boolean equals(Object o) { if (this == o) { @@ -95,14 +86,6 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> { return state == that.state && Objects.equals(action, that.action) && Objects.equals(requestedTime, that.requestedTime); } - public State getState() { - return state; - } - - public String getCompletionTime() { - return completionTime; - } - @Override public int hashCode() { return Objects.hash(state, action, requestedTime); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java index 19938f7cd7cc..267e3f158574 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/LSMTimeline.java @@ -29,9 +29,8 @@ import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathFilter; import org.apache.hudi.storage.StoragePathInfo; +import lombok.extern.slf4j.Slf4j; import org.apache.avro.Schema; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.FileNotFoundException; import java.io.IOException; @@ -103,8 +102,8 @@ import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes; * <p><h3>Instants TTL</h3></p> * The timeline reader only reads instants of last limited days. We will by default skip the instants from LSM timeline that are generated long time ago. */ +@Slf4j public class LSMTimeline { - private static final Logger LOG = LoggerFactory.getLogger(LSMTimeline.class); public static final int LSM_TIMELINE_INSTANT_VERSION_1 = 1; @@ -158,7 +157,7 @@ public class LSMTimeline { } } catch (Exception e) { // fallback to manifest file listing. - LOG.warn("Error reading version file {}", versionFilePath, e); + log.warn("Error reading version file {}", versionFilePath, e); } return allSnapshotVersions(metaClient, archivePath).stream().max(Integer::compareTo).orElse(-1); @@ -176,7 +175,7 @@ public class LSMTimeline { .map(LSMTimeline::getManifestVersion) .collect(Collectors.toList()); } catch (FileNotFoundException ex) { - LOG.debug("Archive path {} does not exist", archivePath); + log.debug("Archive path {} does not exist", archivePath); return Collections.emptyList(); } } @@ -256,7 +255,7 @@ public class LSMTimeline { } } catch (NumberFormatException e) { // log and ignore any format warnings - LOG.warn("error getting file layout for archived file: {}", fileName, e); + log.warn("error getting file layout for archived file: {}", fileName, e); } // return default value in case of any errors diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java index e28af75a1ef7..8ea9ccc03fb9 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java @@ -40,10 +40,9 @@ import org.apache.hudi.common.util.ValidationUtils; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; +import lombok.extern.slf4j.Slf4j; import org.apache.avro.generic.GenericRecord; import org.apache.avro.specific.SpecificRecordBase; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.ByteArrayInputStream; import java.io.IOException; @@ -52,8 +51,8 @@ import java.nio.ByteBuffer; /** * Helper class to convert between different action related payloads and {@link HoodieArchivedMetaEntry}. */ +@Slf4j public class MetadataConversionUtils { - private static final Logger LOG = LoggerFactory.getLogger(MetadataConversionUtils.class); public static HoodieArchivedMetaEntry createMetaWrapper(HoodieInstant hoodieInstant, HoodieTableMetaClient metaClient) { try { @@ -387,17 +386,17 @@ public class MetadataConversionUtils { if (metadata instanceof HoodieReplaceCommitMetadata) { HoodieReplaceCommitMetadata hoodieCommitMetadata = (HoodieReplaceCommitMetadata) metadata; if (hoodieCommitMetadata.getPartitionToWriteStats().containsKey(null)) { - LOG.info("partition path is null for {}", hoodieCommitMetadata.getPartitionToWriteStats().get(null)); + log.info("partition path is null for {}", hoodieCommitMetadata.getPartitionToWriteStats().get(null)); hoodieCommitMetadata.getPartitionToWriteStats().remove(null); } if (hoodieCommitMetadata.getPartitionToReplaceFileIds().containsKey(null)) { - LOG.info("partition path is null for {}", hoodieCommitMetadata.getPartitionToReplaceFileIds().get(null)); + log.info("partition path is null for {}", hoodieCommitMetadata.getPartitionToReplaceFileIds().get(null)); hoodieCommitMetadata.getPartitionToReplaceFileIds().remove(null); } } else if (metadata instanceof HoodieCommitMetadata) { HoodieCommitMetadata hoodieCommitMetadata = (HoodieCommitMetadata) metadata; if (hoodieCommitMetadata.getPartitionToWriteStats().containsKey(null)) { - LOG.info("partition path is null for {}", hoodieCommitMetadata.getPartitionToWriteStats().get(null)); + log.info("partition path is null for {}", hoodieCommitMetadata.getPartitionToWriteStats().get(null)); hoodieCommitMetadata.getPartitionToWriteStats().remove(null); } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimeGeneratorBase.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimeGeneratorBase.java index b616640a9029..18d187e2fdf1 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimeGeneratorBase.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimeGeneratorBase.java @@ -26,8 +26,7 @@ import org.apache.hudi.common.util.RetryHelper; import org.apache.hudi.exception.HoodieLockException; import org.apache.hudi.storage.StorageConfiguration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import java.io.Serializable; import java.util.Arrays; @@ -42,10 +41,9 @@ import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_WAIT_ /** * Base time generator facility that maintains lock-related utilities. */ +@Slf4j public abstract class TimeGeneratorBase implements TimeGenerator, Serializable { - private static final Logger LOG = LoggerFactory.getLogger(TimeGeneratorBase.class); - /** * The lock provider. */ @@ -87,7 +85,7 @@ public abstract class TimeGeneratorBase implements TimeGenerator, Serializable { synchronized (this) { if (lockProvider == null) { String lockProviderClass = lockConfiguration.getConfig().getString("hoodie.write.lock.provider"); - LOG.info("LockProvider for TimeGenerator: {}", lockProviderClass); + log.info("LockProvider for TimeGenerator: {}", lockProviderClass); lockProvider = (LockProvider<?>) ReflectionUtils.loadClass(lockProviderClass, new Class<?>[] {LockConfiguration.class, StorageConfiguration.class}, lockConfiguration, storageConf); @@ -121,10 +119,10 @@ public abstract class TimeGeneratorBase implements TimeGenerator, Serializable { if (lockProvider != null) { lockProvider.close(); lockProvider = null; - LOG.info("Released the connection of the timeGenerator lock"); + log.info("Released the connection of the timeGenerator lock"); } } catch (Exception e) { - LOG.info("Unable to release the connection of the timeGenerator lock"); + log.info("Unable to release the connection of the timeGenerator lock"); } } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java index 502565dd0221..a0159dd2bc3e 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineDiffHelper.java @@ -23,8 +23,12 @@ import org.apache.hudi.common.table.timeline.HoodieInstant.State; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.Pair; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; import java.util.ArrayList; import java.util.List; @@ -37,13 +41,10 @@ import static org.apache.hudi.common.table.timeline.InstantComparison.compareTim /** * A helper class used to diff timeline. */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +@Slf4j public class TimelineDiffHelper { - private static final Logger LOG = LoggerFactory.getLogger(TimelineDiffHelper.class); - - private TimelineDiffHelper() { - } - public static TimelineDiffResult getNewInstantsForIncrementalSync(HoodieTableMetaClient metaClient, HoodieTimeline oldTimeline, HoodieTimeline newTimeline) { @@ -72,7 +73,7 @@ public class TimelineDiffHelper { if (!lostPendingCompactions.isEmpty()) { // If a compaction is unscheduled, fall back to complete refresh of fs view since some log files could have been // moved. Its unsafe to incrementally sync in that case. - LOG.warn("Some pending compactions are no longer in new timeline (unscheduled ?). They are: {}", lostPendingCompactions); + log.warn("Some pending compactions are no longer in new timeline (unscheduled ?). They are: {}", lostPendingCompactions); return TimelineDiffResult.UNSAFE_SYNC_RESULT; } List<HoodieInstant> finishedCompactionInstants = compactionInstants.stream() @@ -91,7 +92,7 @@ public class TimelineDiffHelper { return new TimelineDiffResult(newInstants, finishedCompactionInstants, finishedOrRemovedLogCompactionInstants, true); } else { // One or more timelines is empty - LOG.warn("One or more timelines is empty"); + log.warn("One or more timelines is empty"); return TimelineDiffResult.UNSAFE_SYNC_RESULT; } } @@ -125,40 +126,19 @@ public class TimelineDiffHelper { /** * A diff result of timeline. */ + @AllArgsConstructor + @Getter public static class TimelineDiffResult { private final List<HoodieInstant> newlySeenInstants; private final List<HoodieInstant> finishedCompactionInstants; private final List<HoodieInstant> finishedOrRemovedLogCompactionInstants; + @Accessors(fluent = true) private final boolean canSyncIncrementally; public static final TimelineDiffResult UNSAFE_SYNC_RESULT = new TimelineDiffResult(null, null, null, false); - public TimelineDiffResult(List<HoodieInstant> newlySeenInstants, List<HoodieInstant> finishedCompactionInstants, - List<HoodieInstant> finishedOrRemovedLogCompactionInstants, boolean canSyncIncrementally) { - this.newlySeenInstants = newlySeenInstants; - this.finishedCompactionInstants = finishedCompactionInstants; - this.finishedOrRemovedLogCompactionInstants = finishedOrRemovedLogCompactionInstants; - this.canSyncIncrementally = canSyncIncrementally; - } - - public List<HoodieInstant> getNewlySeenInstants() { - return newlySeenInstants; - } - - public List<HoodieInstant> getFinishedCompactionInstants() { - return finishedCompactionInstants; - } - - public List<HoodieInstant> getFinishedOrRemovedLogCompactionInstants() { - return finishedOrRemovedLogCompactionInstants; - } - - public boolean canSyncIncrementally() { - return canSyncIncrementally; - } - @Override public String toString() { return "TimelineDiffResult{" diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineLayout.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineLayout.java index d4ed4b53c916..83ce1f20d2c2 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineLayout.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineLayout.java @@ -34,6 +34,8 @@ import org.apache.hudi.common.table.timeline.versioning.v2.TimelinePathProviderV import org.apache.hudi.common.table.timeline.versioning.v2.TimelineV2Factory; import org.apache.hudi.common.util.collection.Pair; +import lombok.Getter; + import java.io.Serializable; import java.util.HashMap; import java.util.Map; @@ -81,44 +83,20 @@ public abstract class TimelineLayout implements Serializable { /** * Table Layout where state transitions are managed by renaming files. */ + @Getter private static class TimelineLayoutV0 extends TimelineLayout { private final InstantGenerator instantGenerator = new InstantGeneratorV1(); private final InstantFileNameGenerator instantFileNameGenerator = new InstantFileNameGeneratorV1(); private final TimelineFactory timelineFactory = new TimelineV1Factory(this); private final InstantComparator instantComparator = new InstantComparatorV1(); - private final InstantFileNameParser fileNameParser = new InstantFileNameParserV2(); + private final InstantFileNameParser instantFileNameParser = new InstantFileNameParserV2(); @Override public Stream<HoodieInstant> filterHoodieInstants(Stream<HoodieInstant> instantStream) { return instantStream; } - @Override - public InstantGenerator getInstantGenerator() { - return instantGenerator; - } - - @Override - public InstantFileNameGenerator getInstantFileNameGenerator() { - return instantFileNameGenerator; - } - - @Override - public TimelineFactory getTimelineFactory() { - return timelineFactory; - } - - @Override - public InstantComparator getInstantComparator() { - return instantComparator; - } - - @Override - public InstantFileNameParser getInstantFileNameParser() { - return fileNameParser; - } - @Override public CommitMetadataSerDe getCommitMetadataSerDe() { return new CommitMetadataSerDeV1(); @@ -157,44 +135,20 @@ public abstract class TimelineLayout implements Serializable { /** * Timeline corresponding to Hudi 1.x */ + @Getter private static class TimelineLayoutV2 extends TimelineLayout { private final InstantGenerator instantGenerator = new InstantGeneratorV2(); private final InstantFileNameGenerator instantFileNameGenerator = new InstantFileNameGeneratorV2(); private final TimelineFactory timelineFactory = new TimelineV2Factory(this); private final InstantComparator instantComparator = new InstantComparatorV2(); - private final InstantFileNameParser fileNameParser = new InstantFileNameParserV2(); + private final InstantFileNameParser instantFileNameParser = new InstantFileNameParserV2(); @Override public Stream<HoodieInstant> filterHoodieInstants(Stream<HoodieInstant> instantStream) { return TimelineLayout.filterHoodieInstantsByLatestState(instantStream, InstantComparatorV2::getComparableAction); } - @Override - public InstantGenerator getInstantGenerator() { - return instantGenerator; - } - - @Override - public InstantFileNameGenerator getInstantFileNameGenerator() { - return instantFileNameGenerator; - } - - @Override - public TimelineFactory getTimelineFactory() { - return timelineFactory; - } - - @Override - public InstantComparator getInstantComparator() { - return instantComparator; - } - - @Override - public InstantFileNameParser getInstantFileNameParser() { - return fileNameParser; - } - @Override public CommitMetadataSerDe getCommitMetadataSerDe() { return new CommitMetadataSerDeV2(); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java index 4b2e6ee55fc6..6c27d1f8e2a6 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java @@ -40,8 +40,7 @@ import org.apache.hudi.storage.HoodieInstantWriter; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import java.io.ByteArrayInputStream; import java.io.FileNotFoundException; @@ -81,6 +80,7 @@ import static org.apache.hudi.common.table.timeline.InstantComparison.compareTim * 1) HiveSync - this can be used to query partitions that changed since previous sync. * 2) Incremental reads - InputFormats can use this API to query */ +@Slf4j public class TimelineUtils { public static final Set<String> NOT_PARSABLE_TIMESTAMPS = new HashSet<String>(3) { { @@ -89,7 +89,6 @@ public class TimelineUtils { add(HoodieTimeline.FULL_BOOTSTRAP_INSTANT_TS); } }; - private static final Logger LOG = LoggerFactory.getLogger(TimelineUtils.class); /** * Returns partitions that have new data strictly after commitTime. @@ -137,7 +136,7 @@ public class TimelineUtils { }); } catch (HoodieIOException e) { if (e.getCause() instanceof FileNotFoundException) { - LOG.warn("Instant {} not found in storage and has been archived", instant, e); + log.warn("Instant {} not found in storage and has been archived", instant, e); } else { throw e; } @@ -264,7 +263,7 @@ public class TimelineUtils { private static Option<String> getMetadataValue(HoodieTableMetaClient metaClient, String extraMetadataKey, HoodieInstant instant) { try { - LOG.info("reading checkpoint info for:" + instant + " key: " + extraMetadataKey); + log.info("reading checkpoint info for:" + instant + " key: " + extraMetadataKey); byte[] contents = metaClient.getCommitsTimeline().getInstantDetails(instant).get(); if (instant.isCompleted()) { if (contents == null || contents.length == 0) { @@ -476,7 +475,7 @@ public class TimelineUtils { "Found hollow commit: '%s'. Adjust config `%s` accordingly if to avoid throwing this exception.", hollowCommitTimestamp, INCREMENTAL_READ_HANDLE_HOLLOW_COMMIT.key())); case BLOCK: - LOG.warn("Found hollow commit '{}'. Config `{}` was set to `{}`: no data will be returned beyond '{}' until it's completed.", + log.warn("Found hollow commit '{}'. Config `{}` was set to `{}`: no data will be returned beyond '{}' until it's completed.", hollowCommitTimestamp, INCREMENTAL_READ_HANDLE_HOLLOW_COMMIT.key(), handlingMode, hollowCommitTimestamp); return completedCommitTimeline.findInstantsBefore(hollowCommitTimestamp); default: @@ -518,7 +517,7 @@ public class TimelineUtils { if (NOT_PARSABLE_TIMESTAMPS.contains(timestamp)) { parsedDate = Option.of(new Date(Integer.parseInt(timestamp))); } else { - LOG.warn("Failed to parse timestamp {}: {}", timestamp, e.getMessage()); + log.warn("Failed to parse timestamp {}: {}", timestamp, e.getMessage()); parsedDate = Option.empty(); } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/TimelineLayoutVersion.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/TimelineLayoutVersion.java index feceb81e33b6..cb2cee9f30ac 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/TimelineLayoutVersion.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/TimelineLayoutVersion.java @@ -20,12 +20,15 @@ package org.apache.hudi.common.table.timeline.versioning; import org.apache.hudi.common.util.ValidationUtils; +import lombok.Getter; + import java.io.Serializable; import java.util.Objects; /** * Metadata Layout Version. Add new version when timeline format changes */ +@Getter public class TimelineLayoutVersion implements Serializable, Comparable<TimelineLayoutVersion> { public static final Integer VERSION_0 = 0; // pre 0.5.1 version format @@ -38,7 +41,6 @@ public class TimelineLayoutVersion implements Serializable, Comparable<TimelineL public static final TimelineLayoutVersion LAYOUT_VERSION_2 = new TimelineLayoutVersion(VERSION_2); public static final TimelineLayoutVersion CURR_LAYOUT_VERSION = LAYOUT_VERSION_2; - private final Integer version; public TimelineLayoutVersion(Integer version) { @@ -56,10 +58,6 @@ public class TimelineLayoutVersion implements Serializable, Comparable<TimelineL return Objects.equals(version, VERSION_0); } - public Integer getVersion() { - return version; - } - @Override public boolean equals(Object o) { if (this == o) { diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ActiveTimelineV1.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ActiveTimelineV1.java index c670404ac769..402bf40fa352 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ActiveTimelineV1.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ActiveTimelineV1.java @@ -47,8 +47,8 @@ import org.apache.hudi.storage.HoodieInstantWriter; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.io.InputStream; @@ -62,6 +62,9 @@ import java.util.stream.Stream; import static org.apache.hudi.common.table.timeline.TimelineUtils.getHoodieInstantWriterOption; +// no-arg constructor is for serialization and de-serialization only; @Deprecated marks it as such +@NoArgsConstructor(onConstructor_ = @Deprecated) +@Slf4j public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTimeline { public static final Set<String> VALID_EXTENSIONS_IN_ACTIVE_TIMELINE = new HashSet<>(Arrays.asList( @@ -77,7 +80,6 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime REQUESTED_INDEX_COMMIT_EXTENSION, INFLIGHT_INDEX_COMMIT_EXTENSION, INDEX_COMMIT_EXTENSION, REQUESTED_SAVE_SCHEMA_ACTION_EXTENSION, INFLIGHT_SAVE_SCHEMA_ACTION_EXTENSION, SAVE_SCHEMA_ACTION_EXTENSION)); - private static final Logger LOG = LoggerFactory.getLogger(ActiveTimelineV1.class); protected HoodieTableMetaClient metaClient; private final InstantFileNameGenerator instantFileNameGenerator = new InstantFileNameGeneratorV1(); @@ -89,7 +91,7 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime this.metaClient = metaClient; // multiple casts will make this lambda serializable - // http://docs.oracle.com/javase/specs/jls/se8/html/jls-15.html#jls-15.16 - LOG.debug("Loaded instants upto : " + lastInstant()); + log.debug("Loaded instants upto : {}", lastInstant()); } public ActiveTimelineV1(HoodieTableMetaClient metaClient) { @@ -100,15 +102,6 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime this(metaClient, Collections.unmodifiableSet(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE), applyLayoutFilter); } - /** - * For serialization and de-serialization only. - * - * @deprecated - */ - @Deprecated - public ActiveTimelineV1() { - } - /** * This method is only used when this object is deserialized in a spark executor. * @@ -126,13 +119,13 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime @Override public void createCompleteInstant(HoodieInstant instant) { - LOG.info("Creating a new complete instant {}", instant); + log.info("Creating a new complete instant {}", instant); createFileInMetaPath(instantFileNameGenerator.getFileName(instant), Option.empty(), false); } @Override public void createNewInstant(HoodieInstant instant) { - LOG.info("Creating a new instant {}", instant); + log.info("Creating a new instant {}", instant); // Create the in-flight file createFileInMetaPath(instantFileNameGenerator.getFileName(instant), Option.empty(), false); } @@ -140,7 +133,7 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime @Override public HoodieInstant createRequestedCommitWithReplaceMetadata(String instantTime, String actionType) { HoodieInstant instant = instantGenerator.createNewInstant(HoodieInstant.State.REQUESTED, actionType, instantTime); - LOG.info("Creating a new instant {}", instant); + log.info("Creating a new instant {}", instant); // Create the request replace file createFileInMetaPath(instantFileNameGenerator.getFileName(instant), Option.of(new HoodieRequestedReplaceMetadata()), false); return instant; @@ -148,12 +141,12 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime @Override public <T> HoodieInstant saveAsComplete(HoodieInstant instant, Option<T> metadata) { - LOG.info("Marking instant complete " + instant); + log.info("Marking instant complete {}", instant); ValidationUtils.checkArgument(instant.isInflight(), "Could not mark an already completed instant as complete again " + instant); HoodieInstant completedInstant = instantGenerator.createNewInstant(HoodieInstant.State.COMPLETED, instant.getAction(), instant.requestedTime()); transitionState(instant, completedInstant, metadata); - LOG.info("Completed {}", instant); + log.info("Completed {}", instant); return completedInstant; } @@ -176,10 +169,10 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime @Override public HoodieInstant revertToInflight(HoodieInstant instant) { - LOG.info("Reverting instant to inflight {}", instant); + log.info("Reverting instant to inflight {}", instant); HoodieInstant inflight = TimelineUtils.getInflightInstant(instant, metaClient); revertCompleteToInflight(instant, inflight); - LOG.info("Reverted {} to inflight {}", instant, inflight); + log.info("Reverted {} to inflight {}", instant, inflight); return inflight; } @@ -216,18 +209,18 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime @Override public void deleteInstantFileIfExists(HoodieInstant instant) { - LOG.info("Deleting instant {}", instant); + log.info("Deleting instant {}", instant); StoragePath commitFilePath = getInstantFileNamePath(instantFileNameGenerator.getFileName(instant)); try { if (metaClient.getStorage().exists(commitFilePath)) { boolean result = metaClient.getStorage().deleteFile(commitFilePath); if (result) { - LOG.info("Removed instant {}", instant); + log.info("Removed instant {}", instant); } else { throw new HoodieIOException("Could not delete instant " + instant + " with path " + commitFilePath); } } else { - LOG.info("The commit {} to remove does not exist", commitFilePath); + log.info("The commit {} to remove does not exist", commitFilePath); } } catch (IOException e) { throw new HoodieIOException("Could not remove commit " + commitFilePath, e); @@ -235,12 +228,12 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime } private void deleteInstantFile(HoodieInstant instant) { - LOG.info("Deleting instant {}", instant); + log.info("Deleting instant {}", instant); StoragePath inFlightCommitFilePath = getInstantFileNamePath(instantFileNameGenerator.getFileName(instant)); try { boolean result = metaClient.getStorage().deleteFile(inFlightCommitFilePath); if (result) { - LOG.info("Removed instant {}", instant); + log.info("Removed instant {}", instant); } else { throw new HoodieIOException("Could not delete instant " + instant + " with path " + inFlightCommitFilePath); } @@ -541,7 +534,7 @@ public class ActiveTimelineV1 extends BaseTimelineV1 implements HoodieActiveTime } else { storage.createImmutableFileInPath(getInstantFileNamePath(instantFileNameGenerator.getFileName(toInstant)), getHoodieInstantWriterOption(this, metadata)); } - LOG.info("Create new file for toInstant ?{}", getInstantFileNamePath(instantFileNameGenerator.getFileName(toInstant))); + log.info("Create new file for toInstant? {}", getInstantFileNamePath(instantFileNameGenerator.getFileName(toInstant))); } } catch (IOException e) { throw new HoodieIOException("Could not complete " + fromInstant, e); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineLoaderV1.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineLoaderV1.java index 8485dc7405d9..d95a8aab0c68 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineLoaderV1.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineLoaderV1.java @@ -39,10 +39,9 @@ import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; +import lombok.extern.slf4j.Slf4j; import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.IndexedRecord; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nullable; @@ -61,13 +60,14 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.StreamSupport; +@Slf4j public class ArchivedTimelineLoaderV1 implements ArchivedTimelineLoader { + private static final String MERGE_ARCHIVE_PLAN_NAME = "mergeArchivePlan"; private static final Pattern ARCHIVE_FILE_PATTERN = Pattern.compile("^\\.commits_\\.archive\\.([0-9]+).*"); private static final String STATE_TRANSITION_TIME = "stateTransitionTime"; private static final String ACTION_TYPE_KEY = "actionType"; - private static final Logger LOG = LoggerFactory.getLogger(ArchivedTimelineLoaderV1.class); @Override public void loadInstants(HoodieTableMetaClient metaClient, @@ -172,7 +172,7 @@ public class ArchivedTimelineLoaderV1 implements ArchivedTimelineLoader { HoodieMergeArchiveFilePlan plan = TimelineMetadataUtils.deserializeAvroMetadataLegacy(FileIOUtils.readDataFromPath(storage, planPath).get(), HoodieMergeArchiveFilePlan.class); String mergedArchiveFileName = plan.getMergedArchiveFileName(); if (!StringUtils.isNullOrEmpty(mergedArchiveFileName) && fs.getPath().getName().equalsIgnoreCase(mergedArchiveFileName)) { - LOG.debug("Catch exception because of reading uncompleted merging archive file {}. Ignore it here.", mergedArchiveFileName); + log.debug("Catch exception because of reading uncompleted merging archive file {}. Ignore it here.", mergedArchiveFileName); continue; } } @@ -207,7 +207,7 @@ public class ArchivedTimelineLoaderV1 implements ArchivedTimelineLoader { } } catch (NumberFormatException e) { // log and ignore any format warnings - LOG.warn("error getting suffix for archived file: {}", f.getPath()); + log.warn("error getting suffix for archived file: {}", f.getPath()); } // return default value in case of any errors diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineV1.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineV1.java index 6fa23ec95c46..aeea6e20c868 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineV1.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/ArchivedTimelineV1.java @@ -28,6 +28,7 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; +import lombok.NoArgsConstructor; import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.IndexedRecord; import org.slf4j.Logger; @@ -53,6 +54,8 @@ import java.util.stream.Collectors; import static org.apache.hudi.common.table.timeline.TimelineUtils.getInputStreamOptionLegacy; +// no-arg constructor is for serialization and de-serialization only +@NoArgsConstructor(onConstructor_ = @Deprecated) public class ArchivedTimelineV1 extends BaseTimelineV1 implements HoodieArchivedTimeline, HoodieInstantReader { private static final String HOODIE_COMMIT_ARCHIVE_LOG_FILE_PREFIX = "commits"; private static final String ACTION_TYPE_KEY = "actionType"; @@ -155,14 +158,6 @@ public class ArchivedTimelineV1 extends BaseTimelineV1 implements HoodieArchived this(metaClient, null, new LogFileFilter(logFiles), state); } - /** - * For serialization and de-serialization only. - * - * @deprecated - */ - public ArchivedTimelineV1() { - } - @Override public HoodieInstantReader getInstantReader() { return this; diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CommitMetadataSerDeV1.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CommitMetadataSerDeV1.java index 24be7d83aedc..52a24dcc9d0b 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CommitMetadataSerDeV1.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CommitMetadataSerDeV1.java @@ -26,9 +26,8 @@ import org.apache.hudi.common.util.JsonUtils; import org.apache.hudi.common.util.Option; import org.apache.hudi.storage.HoodieInstantWriter; +import lombok.extern.slf4j.Slf4j; import org.apache.avro.specific.SpecificRecordBase; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.InputStream; @@ -37,8 +36,8 @@ import java.util.function.BooleanSupplier; import static org.apache.hudi.common.table.timeline.MetadataConversionUtils.removeNullKeyFromMapMembersForCommitMetadata; import static org.apache.hudi.common.table.timeline.TimelineMetadataUtils.deserializeAvroMetadata; +@Slf4j public class CommitMetadataSerDeV1 implements CommitMetadataSerDe { - private static final Logger LOG = LoggerFactory.getLogger(CommitMetadataSerDeV1.class); @Override public <T> T deserialize(HoodieInstant instant, InputStream inputStream, BooleanSupplier isEmptyInstant, Class<T> clazz) throws IOException { diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java index f99662780176..18aeae515ec3 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java @@ -28,6 +28,8 @@ import org.apache.hudi.common.table.timeline.InstantComparison; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.VisibleForTesting; +import lombok.Getter; + import java.io.Serializable; import java.time.Instant; import java.util.Date; @@ -40,6 +42,7 @@ import static org.apache.hudi.common.table.timeline.InstantComparison.GREATER_TH import static org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN; public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Serializable { + private static final long serialVersionUID = 1L; private static final long MILLI_SECONDS_IN_THREE_DAYS = 3 * 24 * 3600 * 1000; @@ -60,6 +63,7 @@ public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Seria * a completion query for t5 would trigger lazy loading with this cursor instant updated to t5. * This sliding window model amortizes redundant loading from different queries. */ + @Getter private final String cursorInstant; /** @@ -229,11 +233,6 @@ public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Seria this.beginToCompletionInstantTimeMap.putIfAbsent(beginInstantTime, completionTime); } - @Override - public String getCursorInstant() { - return cursorInstant; - } - @Override public boolean isEmptyTable() { return this.beginToCompletionInstantTimeMap.isEmpty(); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/ActiveTimelineV2.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/ActiveTimelineV2.java index 69e36596e48d..7a45582bad60 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/ActiveTimelineV2.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/ActiveTimelineV2.java @@ -50,8 +50,8 @@ import org.apache.hudi.storage.HoodieInstantWriter; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.io.InputStream; @@ -67,6 +67,9 @@ import java.util.stream.Stream; import static org.apache.hudi.common.table.timeline.TimelineUtils.getHoodieInstantWriterOption; +// no-arg constructor is for serialization and de-serialization only; @Deprecated marks it as such +@NoArgsConstructor(onConstructor_ = @Deprecated) +@Slf4j public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTimeline { public static final Set<String> VALID_EXTENSIONS_IN_ACTIVE_TIMELINE = new HashSet<>(Arrays.asList( @@ -82,8 +85,6 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime REQUESTED_INDEX_COMMIT_EXTENSION, INFLIGHT_INDEX_COMMIT_EXTENSION, INDEX_COMMIT_EXTENSION, REQUESTED_SAVE_SCHEMA_ACTION_EXTENSION, INFLIGHT_SAVE_SCHEMA_ACTION_EXTENSION, SAVE_SCHEMA_ACTION_EXTENSION, REQUESTED_CLUSTERING_COMMIT_EXTENSION, INFLIGHT_CLUSTERING_COMMIT_EXTENSION)); - - private static final Logger LOG = LoggerFactory.getLogger(ActiveTimelineV2.class); protected HoodieTableMetaClient metaClient; private final InstantFileNameGenerator instantFileNameGenerator = new InstantFileNameGeneratorV2(); @@ -95,7 +96,7 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime this.metaClient = metaClient; // multiple casts will make this lambda serializable - // http://docs.oracle.com/javase/specs/jls/se8/html/jls-15.html#jls-15.16 - LOG.debug("Loaded instants upto : {}", lastInstant()); + log.debug("Loaded instants upto: {}", lastInstant()); } public ActiveTimelineV2(HoodieTableMetaClient metaClient) { @@ -106,15 +107,6 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime this(metaClient, Collections.unmodifiableSet(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE), applyLayoutFilter); } - /** - * For serialization and de-serialization only. - * - * @deprecated - */ - @Deprecated - public ActiveTimelineV2() { - } - /** * This method is only used when this object is deserialized in a spark executor. * @@ -132,13 +124,13 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime @Override public void createCompleteInstant(HoodieInstant instant) { - LOG.info("Creating a new complete instant " + instant); + log.info("Creating a new complete instant {}", instant); createCompleteFileInMetaPath(true, instant, Option.empty()); } @Override public void createNewInstant(HoodieInstant instant) { - LOG.info("Creating a new instant " + instant); + log.info("Creating a new instant: {}", instant); ValidationUtils.checkArgument(!instant.isCompleted()); createFileInMetaPath(instantFileNameGenerator.getFileName(instant), Option.empty(), false); } @@ -146,7 +138,7 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime @Override public HoodieInstant createRequestedCommitWithReplaceMetadata(String instantTime, String actionType) { HoodieInstant instant = instantGenerator.createNewInstant(HoodieInstant.State.REQUESTED, actionType, instantTime); - LOG.info("Creating a new instant " + instant); + log.info("Creating a new instant: {}", instant); // Create the request replace file createFileInMetaPath(instantFileNameGenerator.getFileName(instant), Option.of(new HoodieRequestedReplaceMetadata()), false); return instant; @@ -164,12 +156,12 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime @Override public <T> HoodieInstant saveAsComplete(boolean shouldLock, HoodieInstant instant, Option<T> metadata, Option<String> completionTimeOpt) { - LOG.info("Marking instant complete {}", instant); + log.info("Marking instant complete {}", instant); ValidationUtils.checkArgument(instant.isInflight(), "Could not mark an already completed instant as complete again " + instant); HoodieInstant commitInstant = instantGenerator.createNewInstant(HoodieInstant.State.COMPLETED, instant.getAction(), instant.requestedTime()); HoodieInstant completedInstant = transitionStateToComplete(shouldLock, instant, commitInstant, metadata, completionTimeOpt); - LOG.info("Completed " + instant); + log.info("Completed {}", instant); return completedInstant; } @@ -182,10 +174,10 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime @Override public HoodieInstant revertToInflight(HoodieInstant instant) { - LOG.info("Reverting instant to inflight {}", instant); + log.info("Reverting instant to inflight {}", instant); HoodieInstant inflight = TimelineUtils.getInflightInstant(instant, metaClient); revertCompleteToInflight(instant, inflight); - LOG.info("Reverted {} to inflight {}", instant, inflight); + log.info("Reverted {} to inflight {}", instant, inflight); return inflight; } @@ -223,18 +215,18 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime @Override public void deleteInstantFileIfExists(HoodieInstant instant) { - LOG.info("Deleting instant {}", instant); + log.info("Deleting instant {}", instant); StoragePath commitFilePath = getInstantFileNamePath(instantFileNameGenerator.getFileName(instant)); try { if (metaClient.getStorage().exists(commitFilePath)) { boolean result = metaClient.getStorage().deleteFile(commitFilePath); if (result) { - LOG.info("Removed instant {}", instant); + log.info("Removed instant {}", instant); } else { throw new HoodieIOException("Could not delete instant " + instant + " with path " + commitFilePath); } } else { - LOG.info("The commit {} to remove does not exist", commitFilePath); + log.info("The commit {} to remove does not exist", commitFilePath); } } catch (IOException e) { throw new HoodieIOException("Could not remove commit " + commitFilePath, e); @@ -242,12 +234,12 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime } protected void deleteInstantFile(HoodieInstant instant) { - LOG.info("Deleting instant {}", instant); + log.info("Deleting instant {}", instant); StoragePath filePath = getInstantFileNamePath(instantFileNameGenerator.getFileName(instant)); try { boolean result = metaClient.getStorage().deleteFile(filePath); if (result) { - LOG.info("Removed instant {}", instant); + log.info("Removed instant {}", instant); } else { throw new HoodieIOException("Could not delete instant " + instant + " with path " + filePath); } @@ -602,7 +594,7 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime } else { storage.createImmutableFileInPath(getInstantFileNamePath(toInstantFileName), getInstantWriter(metadata)); } - LOG.info("Create new file for toInstant ?{}", getInstantFileNamePath(toInstantFileName)); + log.info("Create new file for toInstant ?{}", getInstantFileNamePath(toInstantFileName)); } } catch (IOException e) { throw new HoodieIOException("Could not complete " + fromInstant, e); @@ -763,7 +755,7 @@ public class ActiveTimelineV2 extends BaseTimelineV2 implements HoodieActiveTime metaClient.getStorage().createImmutableFileInPath(fullPath, writerOption); } completionTimeRef.set(completionTime); - LOG.info("Created new file for toInstant: {}", fullPath); + log.info("Created new file for toInstant: {}", fullPath); }); return completionTimeRef.get(); } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java index e773c40692ee..516b4ad4cedd 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java @@ -29,6 +29,7 @@ import org.apache.hudi.common.table.timeline.InstantComparison; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.VisibleForTesting; +import lombok.Getter; import org.apache.avro.generic.GenericRecord; import java.io.Serializable; @@ -73,6 +74,7 @@ public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Seria * a completion query for t5 would trigger lazy loading with this cursor instant updated to t5. * This sliding window model amortizes redundant loading from different queries. */ + @Getter private volatile String cursorInstant; /** @@ -313,11 +315,6 @@ public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Seria this.instantTimeToCompletionTimeMap.putIfAbsent(beginInstantTime, completionTime); } - @Override - public String getCursorInstant() { - return cursorInstant; - } - @Override public boolean isEmptyTable() { return this.instantTimeToCompletionTimeMap.isEmpty(); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java index 9c4debe19365..c9468f152fbd 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java @@ -50,8 +50,8 @@ import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import java.io.FileNotFoundException; import java.io.IOException; @@ -93,9 +93,9 @@ import static org.apache.hudi.common.table.timeline.InstantComparison.compareTim * </ul> * The actual mechanism of fetching file slices from different view storages is delegated to sub-classes. */ +@Slf4j public abstract class AbstractTableFileSystemView implements SyncableFileSystemView, Serializable { - private static final Logger LOG = LoggerFactory.getLogger(AbstractTableFileSystemView.class); protected final HoodieTableMetadata tableMetadata; protected HoodieTableMetaClient metaClient; @@ -104,13 +104,14 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV // This is the commits timeline that will be visible for all views extending this view // This is nothing but the write timeline, which contains both ingestion and compaction(major and minor) writers. + @Getter private HoodieTimeline visibleCommitsAndCompactionTimeline; // Used to concurrently load and populate partition views private final ConcurrentHashMap<String, Boolean> addedPartitions = new ConcurrentHashMap<>(4096); // Sampling logger for replaced file groups read logs (log at INFO once every 5 times) - private final SamplingLogger replacedFileGroupsReadSamplingLogger = new SamplingLogger(LOG, 5); + private final SamplingLogger replacedFileGroupsReadSamplingLogger = new SamplingLogger(log, 5); // Locks to control concurrency. Sync operations use write-lock blocking all fetch operations. // For the common-case, we allow concurrent read of single or multiple partitions @@ -199,7 +200,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV if (!isPartitionAvailableInStore(partition)) { if (bootstrapIndex.useIndex()) { try (BootstrapIndex.IndexReader reader = bootstrapIndex.createReader()) { - LOG.info("Bootstrap Index available for partition {}", partition); + log.info("Bootstrap Index available for partition {}", partition); List<BootstrapFileMapping> sourceFileMappings = reader.getSourceFileMappingForPartition(partition); addBootstrapBaseFileMapping(sourceFileMappings.stream() @@ -211,7 +212,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV } }); long storePartitionsTs = timer.endTimer(); - LOG.debug("addFilesToView: NumFiles={}, NumFileGroups={}, FileGroupsCreationTime={}, StoreTimeTaken={}", + log.debug("addFilesToView: NumFiles={}, NumFileGroups={}, FileGroupsCreationTime={}, StoreTimeTaken={}", statuses.size(), fileGroups.size(), fgBuildTimeTakenMs, storePartitionsTs); return fileGroups; } @@ -289,7 +290,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV if (ex.getIOException() instanceof FileNotFoundException) { // Replace instant could be deleted by archive and FileNotFoundException could be threw during getInstantDetails function // So that we need to catch the FileNotFoundException here and continue - LOG.warn(ex.getMessage()); + log.warn(ex.getMessage()); return Stream.empty(); } else { throw ex; @@ -405,9 +406,9 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV try { // For metadata table, log at DEBUG. For data table, log at INFO. if (metaClient.isMetadataTable()) { - LOG.debug("Building file system view for {} partition(s)", partitionSet.size()); + log.debug("Building file system view for {} partition(s)", partitionSet.size()); } else { - LOG.info("Building file system view for {} partition(s)", partitionSet.size()); + log.info("Building file system view for {} partition(s)", partitionSet.size()); } // Pairs of relative partition path and absolute partition path @@ -419,20 +420,20 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV Map<Pair<String, StoragePath>, List<StoragePathInfo>> pathInfoMap = tableMetadata.listPartitions(absolutePartitionPathList); long endLsTs = System.currentTimeMillis(); - LOG.debug("Time taken to list partitions {} ={}", partitionSet, (endLsTs - beginLsTs)); + log.debug("Time taken to list partitions {} ={}", partitionSet, (endLsTs - beginLsTs)); pathInfoMap.forEach((partitionPair, statuses) -> { String relativePartitionStr = partitionPair.getLeft(); List<HoodieFileGroup> groups = addFilesToView(relativePartitionStr, statuses); if (groups.isEmpty()) { storePartitionView(relativePartitionStr, Collections.emptyList()); } - LOG.debug("#files found in partition ({}) ={}", relativePartitionStr, statuses.size()); + log.debug("#files found in partition ({}) ={}", relativePartitionStr, statuses.size()); }); } catch (IOException e) { throw new HoodieIOException("Failed to list base files in partitions " + partitionSet, e); } long endTs = System.currentTimeMillis(); - LOG.debug("Time to load partition {} ={}", partitionSet, (endTs - beginTs)); + log.debug("Time to load partition {} ={}", partitionSet, (endTs - beginTs)); } partitionSet.forEach(partition -> @@ -451,7 +452,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV long beginLsTs = System.currentTimeMillis(); List<StoragePathInfo> pathInfoList = tableMetadata.getAllFilesInPartition(partitionPath); long endLsTs = System.currentTimeMillis(); - LOG.debug( + log.debug( "#files found in partition ({}}) = {}, Time taken ={}", relativePartitionPath, pathInfoList.size(), (endLsTs - beginLsTs)); return pathInfoList; } @@ -473,9 +474,9 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV try { // For metadata table, log at DEBUG. For data table, log at INFO. if (metaClient.isMetadataTable()) { - LOG.debug("Building file system view for partition ({})", partitionPathStr); + log.debug("Building file system view for partition ({})", partitionPathStr); } else { - LOG.info("Building file system view for partition ({})", partitionPathStr); + log.info("Building file system view for partition ({})", partitionPathStr); } List<HoodieFileGroup> groups = addFilesToView(partitionPathStr, getAllFilesInPartition(partitionPathStr)); if (groups.isEmpty()) { @@ -485,10 +486,10 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV throw new HoodieIOException("Failed to list base files in partition " + partitionPathStr, e); } } else { - LOG.debug("View already built for Partition :{}", partitionPathStr); + log.debug("View already built for Partition :{}", partitionPathStr); } long endTs = System.currentTimeMillis(); - LOG.debug("Time to load partition ({}) ={}", partitionPathStr, (endTs - beginTs)); + log.debug("Time to load partition ({}) ={}", partitionPathStr, (endTs - beginTs)); return true; }); } @@ -580,7 +581,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV */ protected Stream<FileSlice> filterBaseFileAfterPendingCompaction(FileSlice fileSlice, boolean includeEmptyFileSlice) { if (isFileSliceAfterPendingCompaction(fileSlice)) { - LOG.debug("File Slice ({}) is in pending compaction", fileSlice); + log.debug("File Slice ({}) is in pending compaction", fileSlice); // Base file is filtered out of the file-slice as the corresponding compaction // instant not completed yet. FileSlice transformed = new FileSlice(fileSlice.getPartitionPath(), fileSlice.getBaseInstantTime(), fileSlice.getFileId()); @@ -606,7 +607,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV .collect(Collectors.toList()); if ((fileSlice.getBaseFile().isPresent() && !committedBaseFile.isPresent()) || committedLogFiles.size() != fileSlice.getLogFileCnt()) { - LOG.debug("File Slice ({}) has uncommitted files.", fileSlice); + log.debug("File Slice ({}) has uncommitted files.", fileSlice); // A file is filtered out of the file-slice if the corresponding // instant has not completed yet. FileSlice transformed = new FileSlice(fileSlice.getPartitionPath(), fileSlice.getBaseInstantTime(), fileSlice.getFileId()); @@ -630,7 +631,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV .filter(logFile -> completionTimeQueryView.isCompleted(logFile.getDeltaCommitTime())) .collect(Collectors.toList()); if (committedLogFiles.size() != fileSlice.getLogFileCnt()) { - LOG.debug("File Slice ({}) has uncommitted log files.", fileSlice); + log.debug("File Slice ({}) has uncommitted log files.", fileSlice); // A file is filtered out of the file-slice if the corresponding // instant has not completed yet. FileSlice transformed = new FileSlice(fileSlice.getPartitionPath(), fileSlice.getBaseInstantTime(), fileSlice.getFileId()); @@ -1206,7 +1207,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV private Map<HoodieFileGroupId, BootstrapBaseFileMapping> getBootstrapBaseFileMappings(String partition) { try (BootstrapIndex.IndexReader reader = bootstrapIndex.createReader()) { - LOG.info("Bootstrap Index available for partition {}", partition); + log.info("Bootstrap Index available for partition {}", partition); List<BootstrapFileMapping> sourceFileMappings = reader.getSourceFileMappingForPartition(partition); return sourceFileMappings.stream() @@ -1742,13 +1743,4 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV writeLock.unlock(); } } - - /** - * Return Only Commits and Compaction timeline for building file-groups. - * - * @return {@code HoodieTimeline} - */ - public HoodieTimeline getVisibleCommitsAndCompactionTimeline() { - return visibleCommitsAndCompactionTimeline; - } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/FileSystemViewManager.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/FileSystemViewManager.java index d1baad849516..c77e3b99152a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/FileSystemViewManager.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/FileSystemViewManager.java @@ -32,8 +32,7 @@ import org.apache.hudi.metadata.FileSystemBackedTableMetadata; import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.storage.StorageConfiguration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import java.util.concurrent.ConcurrentHashMap; @@ -58,8 +57,8 @@ import java.util.concurrent.ConcurrentHashMap; * view. FileSystemViewManager uses a factory to construct specific implementation of file-system view and passes it to * clients for querying. */ +@Slf4j public class FileSystemViewManager { - private static final Logger LOG = LoggerFactory.getLogger(FileSystemViewManager.class); private static final String HOODIE_METASERVER_FILE_SYSTEM_VIEW_CLASS = "org.apache.hudi.common.table.view.HoodieMetaserverFileSystemView"; @@ -142,7 +141,7 @@ public class FileSystemViewManager { private static RocksDbBasedFileSystemView createRocksDBBasedFileSystemView(HoodieEngineContext engineContext, FileSystemViewStorageConfig viewConf, HoodieTableMetaClient metaClient, boolean metadataTableEnabled, SerializableFunctionUnchecked<HoodieTableMetaClient, HoodieTableMetadata> metadataCreator) { - LOG.info("Creating RocksDB based view for basePath {}.", metaClient.getBasePath()); + log.info("Creating RocksDB based view for basePath {}.", metaClient.getBasePath()); HoodieTimeline timeline = metaClient.getActiveTimeline().filterCompletedAndCompactionInstants(); HoodieTableMetadata tableMetadata = getTableMetadata(engineContext, metaClient, metadataTableEnabled, metadataCreator); return new RocksDbBasedFileSystemView(tableMetadata, metaClient, timeline, viewConf); @@ -159,7 +158,7 @@ public class FileSystemViewManager { HoodieTableMetaClient metaClient, HoodieCommonConfig commonConfig, boolean metadataTableEnabled, SerializableFunctionUnchecked<HoodieTableMetaClient, HoodieTableMetadata> metadataCreator) { - LOG.info("Creating SpillableMap based view for basePath {}.", metaClient.getBasePath()); + log.info("Creating SpillableMap based view for basePath {}.", metaClient.getBasePath()); HoodieTimeline timeline = metaClient.getActiveTimeline().filterCompletedAndCompactionInstants(); HoodieTableMetadata tableMetadata = getTableMetadata(engineContext, metaClient, metadataTableEnabled, metadataCreator); return new SpillableMapBasedFileSystemView(tableMetadata, metaClient, timeline, viewConf, commonConfig); @@ -171,7 +170,7 @@ public class FileSystemViewManager { private static HoodieTableFileSystemView createInMemoryFileSystemView(HoodieEngineContext engineContext, FileSystemViewStorageConfig viewConf, HoodieTableMetaClient metaClient, boolean metadataTableEnabled, SerializableFunctionUnchecked<HoodieTableMetaClient, HoodieTableMetadata> metadataCreator) { - LOG.info("Creating InMemory based view for basePath {}.", metaClient.getBasePath()); + log.info("Creating InMemory based view for basePath {}.", metaClient.getBasePath()); HoodieTimeline timeline = metaClient.getActiveTimeline().filterCompletedAndCompactionInstants(); HoodieTableMetadata tableMetadata = getTableMetadata(engineContext, metaClient, metadataTableEnabled, metadataCreator); if (metaClient.getMetaserverConfig().isMetaserverEnabled()) { @@ -204,7 +203,7 @@ public class FileSystemViewManager { HoodieTableMetaClient metaClient, HoodieMetadataConfig metadataConfig, HoodieTimeline timeline) { - LOG.info("Creating InMemory based view for basePath {}.", metaClient.getBasePath()); + log.info("Creating InMemory based view for basePath {}.", metaClient.getBasePath()); HoodieTableMetadata tableMetadata = getTableMetadata(engineContext, metaClient, metadataConfig.isEnabled(), unused -> metaClient.getTableFormat().getMetadataFactory().create(engineContext, metaClient.getStorage(), metadataConfig, metaClient.getBasePath().toString())); @@ -225,7 +224,7 @@ public class FileSystemViewManager { */ private static RemoteHoodieTableFileSystemView createRemoteFileSystemView(FileSystemViewStorageConfig viewConf, HoodieTableMetaClient metaClient) { - LOG.info("Creating remote view for basePath {}. Server={}:{}, Timeout={}", metaClient.getBasePath(), + log.info("Creating remote view for basePath {}. Server={}:{}, Timeout={}", metaClient.getBasePath(), viewConf.getRemoteViewServerHost(), viewConf.getRemoteViewServerPort(), viewConf.getRemoteTimelineClientTimeoutSecs()); return new RemoteHoodieTableFileSystemView(metaClient, viewConf); } @@ -254,27 +253,27 @@ public class FileSystemViewManager { final FileSystemViewStorageConfig config, final HoodieCommonConfig commonConfig, final SerializableFunctionUnchecked<HoodieTableMetaClient, HoodieTableMetadata> metadataCreator) { - LOG.info("Creating View Manager with storage type {}.", config.getStorageType()); + log.info("Creating View Manager with storage type {}.", config.getStorageType()); boolean metadataTableEnabled = metadataConfig.isEnabled(); switch (config.getStorageType()) { case EMBEDDED_KV_STORE: - LOG.debug("Creating embedded rocks-db based Table View"); + log.debug("Creating embedded rocks-db based Table View"); return new FileSystemViewManager(context, config, (metaClient, viewConf) -> createRocksDBBasedFileSystemView(context, viewConf, metaClient, metadataTableEnabled, metadataCreator)); case SPILLABLE_DISK: - LOG.debug("Creating Spillable Disk based Table View"); + log.debug("Creating Spillable Disk based Table View"); return new FileSystemViewManager(context, config, (metaClient, viewConf) -> createSpillableMapBasedFileSystemView(context, viewConf, metaClient, commonConfig, metadataTableEnabled, metadataCreator)); case MEMORY: - LOG.debug("Creating in-memory based Table View"); + log.debug("Creating in-memory based Table View"); return new FileSystemViewManager(context, config, (metaClient, viewConfig) -> createInMemoryFileSystemView(context, viewConfig, metaClient, metadataTableEnabled, metadataCreator)); case REMOTE_ONLY: - LOG.debug("Creating remote only table view"); + log.debug("Creating remote only table view"); return new FileSystemViewManager(context, config, (metaClient, viewConfig) -> createRemoteFileSystemView(viewConfig, metaClient)); case REMOTE_FIRST: - LOG.debug("Creating remote first table view"); + log.debug("Creating remote first table view"); return new FileSystemViewManager(context, config, (metaClient, viewConfig) -> { RemoteHoodieTableFileSystemView remoteFileSystemView = createRemoteFileSystemView(viewConfig, metaClient); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java index 10d8879f8233..cb1aa5723a0a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java @@ -33,8 +33,8 @@ import org.apache.hudi.metadata.FileSystemBackedTableMetadata; import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.storage.StoragePathInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.util.ArrayList; @@ -52,10 +52,9 @@ import java.util.stream.Stream; * @see TableFileSystemView * @since 0.3.0 */ +@Slf4j public class HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystemView { - private static final Logger LOG = LoggerFactory.getLogger(HoodieTableFileSystemView.class); - //TODO: [HUDI-6249] change the maps below to implement ConcurrentMap // mapping from partition paths to file groups contained within them @@ -89,6 +88,7 @@ public class HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystem /** * Flag to determine if closed. */ + @Getter private boolean closed = false; HoodieTableFileSystemView(HoodieTableMetadata tableMetadata, boolean enableIncrementalTimelineSync) { @@ -402,7 +402,7 @@ public class HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystem @Override protected void storePartitionView(String partitionPath, List<HoodieFileGroup> fileGroups) { - LOG.debug("Adding file-groups for partition :{}, #FileGroups={}", partitionPath, fileGroups.size()); + log.debug("Adding file-groups for partition :{}, #FileGroups={}", partitionPath, fileGroups.size()); List<HoodieFileGroup> newList = new ArrayList<>(fileGroups); partitionToFileGroupsMap.put(partitionPath, newList); } @@ -453,9 +453,4 @@ public class HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystem public void close() { super.close(); } - - @Override - public boolean isClosed() { - return closed; - } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java index c56ccbe97eb6..997fbdea224b 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java @@ -47,8 +47,7 @@ import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.util.List; @@ -59,10 +58,9 @@ import java.util.stream.Collectors; /** * Adds the capability to incrementally sync the changes to file-system view as and when new instants gets completed. */ +@Slf4j public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTableFileSystemView { - private static final Logger LOG = LoggerFactory.getLogger(IncrementalTimelineSyncFileSystemView.class); - // Allows incremental Timeline syncing private final boolean incrementalTimelineSyncEnabled; @@ -98,19 +96,19 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl if (incrementalTimelineSyncEnabled) { TimelineDiffResult diffResult = TimelineDiffHelper.getNewInstantsForIncrementalSync(metaClient, oldTimeline, newTimeline); if (diffResult.canSyncIncrementally()) { - LOG.info("Doing incremental sync"); + log.info("Doing incremental sync"); // need to refresh the completion time query view // before amending existing file groups. refreshCompletionTimeQueryView(); runIncrementalSync(newTimeline, diffResult); - LOG.info("Finished incremental sync"); + log.info("Finished incremental sync"); // Reset timeline to latest refreshTimeline(newTimeline); return; } } } catch (Exception ioe) { - LOG.error("Got exception trying to perform incremental sync. Reverting to complete sync", ioe); + log.error("Got exception trying to perform incremental sync. Reverting to complete sync", ioe); } clear(); // Initialize with new Hoodie timeline. @@ -125,7 +123,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl */ private void runIncrementalSync(HoodieTimeline timeline, TimelineDiffResult diffResult) { - LOG.info("Timeline Diff Result is :{}", diffResult); + log.info("Timeline Diff Result is :{}", diffResult); // First remove pending compaction instants which were completed diffResult.getFinishedCompactionInstants().stream().forEach(instant -> { @@ -180,7 +178,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Compaction Instant to be removed */ private void removePendingCompactionInstant(HoodieInstant instant) throws IOException { - LOG.info("Removing completed compaction instant ({})", instant); + log.info("Removing completed compaction instant ({})", instant); HoodieCompactionPlan plan = CompactionUtils.getCompactionPlan(metaClient, instant.requestedTime()); removePendingCompactionOperations(CompactionUtils.getPendingCompactionOperations(instant, plan) .map(instantPair -> Pair.of(instantPair.getValue().getKey(), @@ -194,7 +192,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Log Compaction Instant to be removed */ private void removePendingLogCompactionInstant(HoodieInstant instant) throws IOException { - LOG.info("Removing completed log compaction instant ({})", instant); + log.info("Removing completed log compaction instant ({})", instant); HoodieCompactionPlan plan = CompactionUtils.getLogCompactionPlan(metaClient, instant.requestedTime()); removePendingLogCompactionOperations(CompactionUtils.getPendingCompactionOperations(instant, plan) .map(instantPair -> Pair.of(instantPair.getValue().getKey(), @@ -208,7 +206,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Compaction Instant */ private void addPendingCompactionInstant(HoodieTimeline timeline, HoodieInstant instant) throws IOException { - LOG.info("Syncing pending compaction instant ({})", instant); + log.info("Syncing pending compaction instant ({})", instant); HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(metaClient, instant.requestedTime()); List<Pair<String, CompactionOperation>> pendingOps = CompactionUtils.getPendingCompactionOperations(instant, compactionPlan) @@ -238,7 +236,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Compaction Instant */ private void addPendingLogCompactionInstant(HoodieInstant instant) throws IOException { - LOG.info("Syncing pending log compaction instant ({})", instant); + log.info("Syncing pending log compaction instant ({})", instant); HoodieCompactionPlan compactionPlan = CompactionUtils.getLogCompactionPlan(metaClient, instant.requestedTime()); List<Pair<String, CompactionOperation>> pendingOps = CompactionUtils.getPendingCompactionOperations(instant, compactionPlan) @@ -257,10 +255,10 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Instant */ private void addCommitInstant(HoodieTimeline timeline, HoodieInstant instant) throws IOException { - LOG.info("Syncing committed instant ({})", instant); + log.info("Syncing committed instant ({})", instant); HoodieCommitMetadata commitMetadata = timeline.readCommitMetadata(instant); updatePartitionWriteFileGroups(commitMetadata.getPartitionToWriteStats(), timeline, instant); - LOG.info("Done Syncing committed instant ({})", instant); + log.info("Done Syncing committed instant ({})", instant); } private void updatePartitionWriteFileGroups(Map<String, List<HoodieWriteStat>> partitionToWriteStats, @@ -269,7 +267,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl partitionToWriteStats.entrySet().stream().forEach(entry -> { String partition = entry.getKey(); if (isPartitionAvailableInStore(partition)) { - LOG.info("Syncing partition ({}) of instant ({})", partition, instant); + log.info("Syncing partition ({}) of instant ({})", partition, instant); List<StoragePathInfo> pathInfoList = entry.getValue().stream() .map(p -> new StoragePathInfo( new StoragePath(String.format("%s/%s", metaClient.getBasePath(), p.getPath())), @@ -279,10 +277,10 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl buildFileGroups(partition, pathInfoList, timeline.filterCompletedAndCompactionInstants(), false); applyDeltaFileSlicesToPartitionView(partition, fileGroups, DeltaApplyMode.ADD); } else { - LOG.warn("Skipping partition ({}) when syncing instant ({}) as it is not loaded", partition, instant); + log.warn("Skipping partition ({}) when syncing instant ({}) as it is not loaded", partition, instant); } }); - LOG.info("Done Syncing committed instant ({})", instant); + log.info("Done Syncing committed instant ({})", instant); } /** @@ -292,7 +290,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Restore Instant */ private void addRestoreInstant(HoodieTimeline timeline, HoodieInstant instant) throws IOException { - LOG.info("Syncing restore instant ({})", instant); + log.info("Syncing restore instant ({})", instant); HoodieRestoreMetadata metadata = timeline.readRestoreMetadata(instant); Map<String, List<Pair<String, String>>> partitionFiles = @@ -312,7 +310,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl .map(HoodieInstantInfo::getCommitTime).collect(Collectors.toSet()); removeReplacedFileIdsAtInstants(rolledbackInstants); } - LOG.info("Done Syncing restore instant ({})", instant); + log.info("Done Syncing restore instant ({})", instant); } /** @@ -322,13 +320,13 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Rollback Instant */ private void addRollbackInstant(HoodieTimeline timeline, HoodieInstant instant) throws IOException { - LOG.info("Syncing rollback instant ({})", instant); + log.info("Syncing rollback instant ({})", instant); HoodieRollbackMetadata metadata = timeline.readRollbackMetadata(instant); metadata.getPartitionMetadata().entrySet().stream().forEach(e -> { removeFileSlicesForPartition(timeline, instant, e.getKey(), e.getValue().getSuccessDeleteFiles()); }); - LOG.info("Done Syncing rollback instant ({})", instant); + log.info("Done Syncing rollback instant ({})", instant); } /** @@ -338,7 +336,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant REPLACE Instant */ private void addReplaceInstant(HoodieTimeline timeline, HoodieInstant instant) throws IOException { - LOG.info("Syncing replace instant ({})", instant); + log.info("Syncing replace instant ({})", instant); HoodieReplaceCommitMetadata replaceMetadata = timeline.readReplaceCommitMetadata(instant); updatePartitionWriteFileGroups(replaceMetadata.getPartitionToWriteStats(), timeline, instant); replaceMetadata.getPartitionToReplaceFileIds().entrySet().stream().forEach(entry -> { @@ -346,10 +344,10 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl Map<HoodieFileGroupId, HoodieInstant> replacedFileIds = entry.getValue().stream() .collect(Collectors.toMap(replaceStat -> new HoodieFileGroupId(partition, replaceStat), replaceStat -> instant)); - LOG.info("For partition ({}) of instant ({}), excluding {} file groups", partition, instant, replacedFileIds.size()); + log.info("For partition ({}) of instant ({}), excluding {} file groups", partition, instant, replacedFileIds.size()); addReplacedFileGroups(replacedFileIds); }); - LOG.info("Done Syncing REPLACE instant ({})", instant); + log.info("Done Syncing REPLACE instant ({})", instant); } /** @@ -360,7 +358,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl * @param instant Clean instant */ private void addCleanInstant(HoodieTimeline timeline, HoodieInstant instant) throws IOException { - LOG.info("Syncing cleaner instant ({})", instant); + log.info("Syncing cleaner instant ({})", instant); HoodieCleanMetadata cleanMetadata = CleanerUtils.getCleanerMetadata(metaClient, instant); cleanMetadata.getPartitionMetadata().entrySet().stream().forEach(entry -> { final StoragePath basePath = metaClient.getBasePath(); @@ -371,13 +369,13 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl .collect(Collectors.toList()); removeFileSlicesForPartition(timeline, instant, entry.getKey(), fullPathList); }); - LOG.info("Done Syncing cleaner instant ({})", instant); + log.info("Done Syncing cleaner instant ({})", instant); } private void removeFileSlicesForPartition(HoodieTimeline timeline, HoodieInstant instant, String partition, List<String> paths) { if (isPartitionAvailableInStore(partition)) { - LOG.info("Removing file slices for partition ({}) for instant ({})", partition, instant); + log.info("Removing file slices for partition ({}) for instant ({})", partition, instant); List<StoragePathInfo> pathInfoList = paths.stream() .map(p -> new StoragePathInfo(new StoragePath(p), 0, false, (short) 0, 0, 0)) .collect(Collectors.toList()); @@ -385,7 +383,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl buildFileGroups(partition, pathInfoList, timeline.filterCompletedAndCompactionInstants(), false); applyDeltaFileSlicesToPartitionView(partition, fileGroups, DeltaApplyMode.REMOVE); } else { - LOG.warn("Skipping partition ({}) when syncing instant ({}) as it is not loaded", partition, instant); + log.warn("Skipping partition ({}) when syncing instant ({}) as it is not loaded", partition, instant); } } @@ -408,7 +406,7 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl protected void applyDeltaFileSlicesToPartitionView(String partition, List<HoodieFileGroup> deltaFileGroups, DeltaApplyMode mode) { if (deltaFileGroups.isEmpty()) { - LOG.info("No delta file groups for partition :{}", partition); + log.info("No delta file groups for partition :{}", partition); return; } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/PriorityBasedFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/PriorityBasedFileSystemView.java index e6d60c13b548..c54c0edcafc3 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/PriorityBasedFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/PriorityBasedFileSystemView.java @@ -34,10 +34,11 @@ import org.apache.hudi.common.util.Functions.Function3; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.Pair; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import org.apache.http.HttpStatus; import org.apache.http.client.HttpResponseException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.Serializable; import java.util.List; @@ -48,11 +49,11 @@ import java.util.stream.Stream; * A file system view which proxies request to a preferred File System View implementation. In case of error, flip all * subsequent calls to a backup file-system view implementation. */ +@Slf4j public class PriorityBasedFileSystemView implements SyncableFileSystemView, Serializable { - private static final Logger LOG = LoggerFactory.getLogger(PriorityBasedFileSystemView.class); - private final transient HoodieEngineContext engineContext; + @Getter(AccessLevel.PACKAGE) private final SyncableFileSystemView preferredView; private final SerializableFunctionUnchecked<HoodieEngineContext, SyncableFileSystemView> secondaryViewCreator; private SyncableFileSystemView secondaryView; @@ -69,7 +70,7 @@ public class PriorityBasedFileSystemView implements SyncableFileSystemView, Seri private <R> R execute(Function0<R> preferredFunction, Function0<R> secondaryFunction) { if (errorOnPreferredView) { - LOG.warn("Routing request to secondary file-system view"); + log.warn("Routing request to secondary file-system view"); return secondaryFunction.apply(); } else { try { @@ -84,7 +85,7 @@ public class PriorityBasedFileSystemView implements SyncableFileSystemView, Seri private <T1, R> R execute(T1 val, Function1<T1, R> preferredFunction, Function1<T1, R> secondaryFunction) { if (errorOnPreferredView) { - LOG.warn("Routing request to secondary file-system view"); + log.warn("Routing request to secondary file-system view"); return secondaryFunction.apply(val); } else { try { @@ -100,7 +101,7 @@ public class PriorityBasedFileSystemView implements SyncableFileSystemView, Seri private <T1, T2, R> R execute(T1 val, T2 val2, Function2<T1, T2, R> preferredFunction, Function2<T1, T2, R> secondaryFunction) { if (errorOnPreferredView) { - LOG.warn("Routing request to secondary file-system view"); + log.warn("Routing request to secondary file-system view"); return secondaryFunction.apply(val, val2); } else { try { @@ -116,7 +117,7 @@ public class PriorityBasedFileSystemView implements SyncableFileSystemView, Seri private <T1, T2, T3, R> R execute(T1 val, T2 val2, T3 val3, Function3<T1, T2, T3, R> preferredFunction, Function3<T1, T2, T3, R> secondaryFunction) { if (errorOnPreferredView) { - LOG.warn("Routing request to secondary file-system view"); + log.warn("Routing request to secondary file-system view"); return secondaryFunction.apply(val, val2, val3); } else { try { @@ -131,9 +132,9 @@ public class PriorityBasedFileSystemView implements SyncableFileSystemView, Seri private void handleRuntimeException(RuntimeException re) { if (re.getCause() instanceof HttpResponseException && ((HttpResponseException)re.getCause()).getStatusCode() == HttpStatus.SC_BAD_REQUEST) { - LOG.warn("Got error running preferred function. Likely due to another concurrent writer in progress. Trying secondary"); + log.warn("Got error running preferred function. Likely due to another concurrent writer in progress. Trying secondary"); } else { - LOG.error("Got error running preferred function. Trying secondary", re); + log.error("Got error running preferred function. Trying secondary", re); } } @@ -354,10 +355,6 @@ public class PriorityBasedFileSystemView implements SyncableFileSystemView, Seri return execute(partitionPath, fileId, preferredView::getLatestFileSlice, (path, fgId) -> getSecondaryView().getLatestFileSlice(path, fgId)); } - SyncableFileSystemView getPreferredView() { - return preferredView; - } - synchronized SyncableFileSystemView getSecondaryView() { if (secondaryView == null) { secondaryView = secondaryViewCreator.apply(engineContext); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java index 4cdcb0968122..431292e88aa2 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java @@ -46,8 +46,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.module.afterburner.AfterburnerModule; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.io.Serializable; @@ -62,6 +61,7 @@ import static org.apache.hudi.timeline.TimelineServiceClient.RequestMethod; /** * A proxy for table file-system view which translates local View API calls to REST calls to remote timeline service. */ +@Slf4j public class RemoteHoodieTableFileSystemView implements SyncableFileSystemView, Serializable { private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper().registerModule(new AfterburnerModule()); @@ -127,8 +127,6 @@ public class RemoteHoodieTableFileSystemView implements SyncableFileSystemView, public static final String INCLUDE_FILES_IN_PENDING_COMPACTION_PARAM = "includependingcompaction"; public static final String MULTI_VALUE_SEPARATOR = ","; - - private static final Logger LOG = LoggerFactory.getLogger(RemoteHoodieTableFileSystemView.class); private static final TypeReference<List<FileSliceDTO>> FILE_SLICE_DTOS_REFERENCE = new TypeReference<List<FileSliceDTO>>() {}; private static final TypeReference<List<FileGroupDTO>> FILE_GROUP_DTOS_REFERENCE = new TypeReference<List<FileGroupDTO>>() {}; private static final TypeReference<Boolean> BOOLEAN_TYPE_REFERENCE = new TypeReference<Boolean>() {}; diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java index 2620a29784fc..fee37f33380b 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java @@ -38,8 +38,9 @@ import org.apache.hudi.exception.HoodieException; import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.storage.StoragePathInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import java.io.Serializable; import java.util.HashMap; @@ -65,16 +66,16 @@ import static org.apache.hudi.common.table.timeline.InstantComparison.compareTim * support view-state preservation across restarts, Hoodie timeline also needs to be stored inorder to detect changes to * timeline across restarts. */ +@Slf4j public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSystemView { - private static final Logger LOG = LoggerFactory.getLogger(RocksDbBasedFileSystemView.class); - private final FileSystemViewStorageConfig config; private final RocksDBSchemaHelper schemaHelper; private RocksDBDAO rocksDB; + @Getter(AccessLevel.PACKAGE) private boolean closed = false; public RocksDbBasedFileSystemView(HoodieTableMetadata tableMetadata, HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline, @@ -96,7 +97,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste protected void init(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline) { schemaHelper.getAllColumnFamilies().forEach(rocksDB::addColumnFamily); super.init(metaClient, visibleActiveTimeline); - LOG.info("Created ROCKSDB based file-system view at {}", config.getRocksdbBasePath()); + log.info("Created ROCKSDB based file-system view at {}", config.getRocksdbBasePath()); } @Override @@ -111,7 +112,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste rocksDB.putInBatch(batch, schemaHelper.getColFamilyForPendingCompaction(), schemaHelper.getKeyForPendingCompactionLookup(opPair.getValue().getFileGroupId()), opPair) ); - LOG.info("Initializing pending compaction operations. Count={}", batch.count()); + log.info("Initializing pending compaction operations. Count={}", batch.count()); }); } @@ -154,7 +155,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste rocksDB.putInBatch(batch, schemaHelper.getColFamilyForPendingLogCompaction(), schemaHelper.getKeyForPendingLogCompactionLookup(opPair.getValue().getFileGroupId()), opPair) ); - LOG.info("Initializing pending Log compaction operations. Count={}", batch.count()); + log.info("Initializing pending Log compaction operations. Count={}", batch.count()); }); } @@ -206,14 +207,14 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste @Override void resetFileGroupsInPendingClustering(Map<HoodieFileGroupId, HoodieInstant> fgIdToInstantMap) { - LOG.info("Resetting file groups in pending clustering to ROCKSDB based file-system view at " + log.info("Resetting file groups in pending clustering to ROCKSDB based file-system view at " + config.getRocksdbBasePath() + ", Total file-groups=" + fgIdToInstantMap.size()); // Delete all replaced file groups rocksDB.prefixDelete(schemaHelper.getColFamilyForFileGroupsInPendingClustering(), "part="); // Now add new entries addFileGroupsInPendingClustering(fgIdToInstantMap.entrySet().stream().map(entry -> Pair.of(entry.getKey(), entry.getValue()))); - LOG.info("Resetting replacedFileGroups to ROCKSDB based file-system view complete"); + log.info("Resetting replacedFileGroups to ROCKSDB based file-system view complete"); } @Override @@ -246,7 +247,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste @Override protected void resetViewState() { - LOG.info("Deleting all rocksdb data associated with table filesystem view"); + log.info("Deleting all rocksdb data associated with table filesystem view"); rocksDB.close(); rocksDB = new RocksDBDAO(metaClient.getBasePath().toString(), config.getRocksdbBasePath()); schemaHelper.getAllColumnFamilies().forEach(rocksDB::addColumnFamily); @@ -277,7 +278,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste @Override protected void storePartitionView(String partitionPath, List<HoodieFileGroup> fileGroups) { - LOG.info("Resetting and adding new partition ({}) to ROCKSDB based file-system view at {}, Total file-groups={}", + log.info("Resetting and adding new partition ({}) to ROCKSDB based file-system view at {}, Total file-groups={}", partitionPath, config.getRocksdbBasePath(), fileGroups.size()); String lookupKey = schemaHelper.getKeyForPartitionLookup(partitionPath); @@ -303,7 +304,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste // record that partition is loaded. rocksDB.put(schemaHelper.getColFamilyForStoredPartitions(), lookupKey, Boolean.TRUE); - LOG.info("Finished adding new partition ({}}) to ROCKSDB based file-system view at {}, Total file-groups={}", + log.info("Finished adding new partition ({}}) to ROCKSDB based file-system view at {}, Total file-groups={}", partitionPath, config.getRocksdbBasePath(), fileGroups.size()); } @@ -322,7 +323,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste } else { FileSlice oldSlice = oldSliceOption.get(); // First remove the file-slice - LOG.info("Removing old Slice in DB. FS={}", oldSlice); + log.info("Removing old Slice in DB. FS={}", oldSlice); rocksDB.deleteInBatch(batch, schemaHelper.getColFamilyForView(), schemaHelper.getKeyForSliceView(fg, oldSlice)); rocksDB.deleteInBatch(batch, schemaHelper.getColFamilyForView(), schemaHelper.getKeyForDataFileView(fg, oldSlice)); @@ -342,11 +343,11 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste deltaLogFiles.entrySet().stream().filter(e -> !logFiles.containsKey(e.getKey())) .forEach(p -> newLogFiles.put(p.getKey(), p.getValue())); newLogFiles.values().forEach(newFileSlice::addLogFile); - LOG.info("Adding back new File Slice after add FS={}", newFileSlice); + log.info("Adding back new File Slice after add FS={}", newFileSlice); return newFileSlice; } case REMOVE: { - LOG.info("Removing old File Slice ={}", fs); + log.info("Removing old File Slice ={}", fs); FileSlice newFileSlice = new FileSlice(oldSlice.getFileGroupId(), oldSlice.getBaseInstantTime()); fs.getBaseFile().orElseGet(() -> { oldSlice.getBaseFile().ifPresent(newFileSlice::setBaseFile); @@ -357,7 +358,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste // Add remaining log files back logFiles.values().forEach(newFileSlice::addLogFile); if (newFileSlice.getBaseFile().isPresent() || (newFileSlice.getLogFiles().count() > 0)) { - LOG.info("Adding back new file-slice after remove FS={}", newFileSlice); + log.info("Adding back new file-slice after remove FS={}", newFileSlice); return newFileSlice; } return null; @@ -406,7 +407,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste rocksDB.putInBatch(batch, schemaHelper.getColFamilyForBootstrapBaseFile(), schemaHelper.getKeyForBootstrapBaseFile(externalBaseFile.getFileGroupId()), externalBaseFile); }); - LOG.info("Initializing external data file mapping. Count={}", batch.count()); + log.info("Initializing external data file mapping. Count={}", batch.count()); }); } @@ -516,14 +517,14 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste @Override protected void resetReplacedFileGroups(final Map<HoodieFileGroupId, HoodieInstant> replacedFileGroups) { - LOG.info("Resetting replacedFileGroups to ROCKSDB based file-system view at " + log.info("Resetting replacedFileGroups to ROCKSDB based file-system view at " + config.getRocksdbBasePath() + ", Total file-groups=" + replacedFileGroups.size()); // Delete all replaced file groups rocksDB.prefixDelete(schemaHelper.getColFamilyForReplacedFileGroups(), "part="); // Now add new entries addReplacedFileGroups(replacedFileGroups); - LOG.info("Resetting replacedFileGroups to ROCKSDB based file-system view complete"); + log.info("Resetting replacedFileGroups to ROCKSDB based file-system view complete"); } @Override @@ -542,7 +543,7 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste }) ); - LOG.info("Finished adding replaced file groups to partition (" + partitionPath + ") to ROCKSDB based view at " + log.info("Finished adding replaced file groups to partition (" + partitionPath + ") to ROCKSDB based view at " + config.getRocksdbBasePath() + ", Total file-groups=" + partitionToReplacedFileGroupsEntry.getValue().size()); }); } @@ -599,20 +600,15 @@ public class RocksDbBasedFileSystemView extends IncrementalTimelineSyncFileSyste public void close() { try { writeLock.lock(); - LOG.info("Closing Rocksdb !!"); + log.info("Closing Rocksdb !!"); closed = true; closeResources(); rocksDB.close(); - LOG.info("Closed Rocksdb !!"); + log.info("Closed Rocksdb !!"); } catch (Exception e) { throw new HoodieException("Unable to close file system view", e); } finally { writeLock.unlock(); } } - - @Override - boolean isClosed() { - return closed; - } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/SpillableMapBasedFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/SpillableMapBasedFileSystemView.java index f531907653b1..37bf6b4b05a2 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/SpillableMapBasedFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/SpillableMapBasedFileSystemView.java @@ -35,8 +35,7 @@ import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.storage.StoragePathInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import java.io.File; import java.io.IOException; @@ -48,10 +47,9 @@ import java.util.stream.Stream; /** * Table FileSystemView implementation where view is stored in spillable disk using fixed memory. */ +@Slf4j public class SpillableMapBasedFileSystemView extends HoodieTableFileSystemView { - private static final Logger LOG = LoggerFactory.getLogger(SpillableMapBasedFileSystemView.class); - private final long maxMemoryForFileGroupMap; private final long maxMemoryForPendingCompaction; private final long maxMemoryForPendingLogCompaction; @@ -75,7 +73,7 @@ public class SpillableMapBasedFileSystemView extends HoodieTableFileSystemView { new File(baseStoreDir).mkdirs(); diskMapType = commonConfig.getSpillableDiskMapType(); isBitCaskDiskMapCompressionEnabled = commonConfig.isBitCaskDiskMapCompressionEnabled(); - LOG.info("Initializing SpillableMapBasedFileSystemView with memory configs: " + log.info("Initializing SpillableMapBasedFileSystemView with memory configs: " + "maxMemoryForFileGroupMap={}, maxMemoryForPendingCompaction={}, maxMemoryForPendingLogCompaction={}, " + "maxMemoryForBootstrapBaseFile={}, maxMemoryForReplaceFileGroups={}, maxMemoryForClusteringFileGroups={}, baseStoreDir={}", maxMemoryForFileGroupMap, maxMemoryForPendingCompaction, maxMemoryForPendingLogCompaction,
