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 50198ac6b433582387e757c8b682f1090233a838 Author: voonhous <[email protected]> AuthorDate: Fri Jun 12 09:53:08 2026 +0800 refactor: Add Lombok annotations to hudi-common module (part 7) (#18944) * refactor: Add Lombok annotations to hudi-common module (part 7) * refactor: Address review comments - HoodieLogBlock: revert lombok @NonNull to javax.annotation.Nonnull; null header/footer is expected on the read path (v2/v3 blocks have no footer), so runtime null checks would NPE on every log block read - IncrementalQueryAnalyzer: make QueryContext constructor private again via @AllArgsConstructor(access = AccessLevel.PRIVATE); fix EMPTY constant to pass Option.empty() instead of null - HoodieTableConfig: use parameterized logging instead of String.format - HoodieTableVersion: document why versionCode uses @Accessors(fluent = true) (cherry picked from commit 612e327b51fbb4fc792dcaeecf47aa86f5ddab89) --- .../hudi/common/table/HoodieTableConfig.java | 24 ++-- .../hudi/common/table/HoodieTableMetaClient.java | 159 +++++++-------------- .../hudi/common/table/HoodieTableVersion.java | 24 ++-- .../hudi/common/table/TableSchemaResolver.java | 12 +- .../hudi/common/table/cdc/HoodieCDCExtractor.java | 6 +- .../hudi/common/table/cdc/HoodieCDCFileSplit.java | 24 +--- .../hudi/common/table/cdc/HoodieCDCOperation.java | 14 +- .../hudi/common/table/checkpoint/Checkpoint.java | 21 ++- .../table/log/AbstractHoodieLogRecordScanner.java | 92 ++++-------- .../table/log/BaseHoodieLogRecordReader.java | 103 +++++-------- .../apache/hudi/common/table/log/FullKeySpec.java | 15 +- .../hudi/common/table/log/HoodieLogFileReader.java | 26 ++-- .../hudi/common/table/log/HoodieLogFormat.java | 12 +- .../common/table/log/HoodieLogFormatReader.java | 8 +- .../table/log/HoodieMergedLogRecordReader.java | 20 +-- .../table/log/HoodieMergedLogRecordScanner.java | 20 +-- .../apache/hudi/common/table/log/InstantRange.java | 17 +-- .../common/table/log/block/HoodieCommandBlock.java | 7 +- .../common/table/log/block/HoodieDataBlock.java | 11 +- .../common/table/log/block/HoodieDeleteBlock.java | 6 +- .../table/log/block/HoodieHFileDataBlock.java | 2 + .../common/table/log/block/HoodieLogBlock.java | 94 +++--------- .../hudi/common/table/read/BufferedRecord.java | 47 ++---- .../table/read/BufferedRecordMergerFactory.java | 7 +- .../hudi/common/table/read/DeleteContext.java | 22 +-- .../table/read/IncrementalQueryAnalyzer.java | 70 +++------ .../buffer/DefaultFileGroupRecordBufferLoader.java | 8 +- .../table/read/buffer/FileGroupRecordBuffer.java | 19 +-- .../buffer/PositionBasedFileGroupRecordBuffer.java | 15 +- .../source/split/HoodieSourceSplitSerializer.java | 12 +- .../hudi/source/TestIncrementalInputSplits.java | 4 +- .../source/TestStreamReadMonitoringFunction.java | 4 +- .../hudi/source/split/TestHoodieSourceSplit.java | 14 +- .../split/TestHoodieSourceSplitSerializer.java | 76 +++++----- 34 files changed, 359 insertions(+), 656 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java index ca775136257c..73703d70d85a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java @@ -65,8 +65,7 @@ import org.apache.hudi.metadata.MetadataPartitionType; 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 javax.annotation.concurrent.Immutable; @@ -119,6 +118,7 @@ import static org.apache.hudi.common.util.StringUtils.nonEmpty; import static org.apache.hudi.common.util.ValidationUtils.checkArgument; @Immutable +@Slf4j @ConfigClassProperty(name = "Hudi Table Basic Configs", groupName = ConfigGroups.Names.TABLE_CONFIG, description = "Configurations of the Hudi Table like type of ingestion, storage formats, hive table name etc." @@ -126,8 +126,6 @@ import static org.apache.hudi.common.util.ValidationUtils.checkArgument; + " initializing a path as hoodie base path and never changes during the lifetime of a hoodie table.") public class HoodieTableConfig extends HoodieConfig { - private static final Logger LOG = LoggerFactory.getLogger(HoodieTableConfig.class); - public static final String HOODIE_PROPERTIES_FILE = "hoodie.properties"; public static final String HOODIE_PROPERTIES_FILE_BACKUP = "hoodie.properties.backup"; public static final String HOODIE_WRITE_TABLE_NAME_KEY = "hoodie.datasource.write.table.name"; @@ -475,7 +473,7 @@ public class HoodieTableConfig extends HoodieConfig { public HoodieTableConfig(HoodieStorage storage, StoragePath metaPath) { super(); StoragePath propertyPath = new StoragePath(metaPath, HOODIE_PROPERTIES_FILE); - LOG.info("Loading table properties from " + propertyPath); + log.info("Loading table properties from {}", propertyPath); try { this.props = fetchConfigs(storage, metaPath, HOODIE_PROPERTIES_FILE, HOODIE_PROPERTIES_FILE_BACKUP, MAX_READ_RETRIES, READ_RETRY_DELAY_MSEC); } catch (IOException e) { @@ -509,7 +507,7 @@ public class HoodieTableConfig extends HoodieConfig { checksum = propsWithChecksum.getProperty(TABLE_CHECKSUM.key()); props.setProperty(TABLE_CHECKSUM.key(), checksum); } - LOG.info("Created properties file at " + propertyPath); + log.info("Created properties file at {}", propertyPath); return checksum; } @@ -556,7 +554,7 @@ public class HoodieTableConfig extends HoodieConfig { propsToDelete.forEach(propToDelete -> props.remove(propToDelete)); checksum = storeProperties(props, out, cfgPath); } - LOG.warn(String.format("%s modified to: %s (at %s)", cfgPath.getName(), props, cfgPath.getParent())); + log.warn("{} modified to: {} (at {})", cfgPath.getName(), props, cfgPath.getParent()); // 5. verify and remove backup. try (InputStream in = storage.open(cfgPath)) { @@ -582,7 +580,7 @@ public class HoodieTableConfig extends HoodieConfig { private static void deleteFile(HoodieStorage storage, StoragePath cfgPath) throws IOException { storage.deleteFile(cfgPath); - LOG.info("Deleted properties file at " + cfgPath); + log.info("Deleted properties file at {}", cfgPath); } /** @@ -685,7 +683,7 @@ public class HoodieTableConfig extends HoodieConfig { boolean valid = tableVersion.greaterThan(firstVersion) || tableVersion.equals(firstVersion); valid = valid || CONFIGS_REQUIRED_FOR_OLDER_VERSIONED_TABLES.contains(configProperty.key()); if (!valid) { - LOG.warn("Table version {} is lower than or equal to config's first version {}. Config {} will be ignored.", + log.warn("Table version {} is lower than or equal to config's first version {}. Config {} will be ignored.", tableVersion, firstVersion, configProperty.key()); } return valid; @@ -1009,12 +1007,12 @@ public class HoodieTableConfig extends HoodieConfig { // Check ordering field name based on record merge mode if (inferredRecordMergeMode == COMMIT_TIME_ORDERING) { if (nonEmpty(orderingFieldNamesAsString)) { - LOG.warn("The ordering field ({}) is specified. COMMIT_TIME_ORDERING " + log.warn("The ordering field ({}) is specified. COMMIT_TIME_ORDERING " + "merge mode does not use ordering field anymore.", orderingFieldNamesAsString); } } else if (inferredRecordMergeMode == EVENT_TIME_ORDERING) { if (isNullOrEmpty(orderingFieldNamesAsString)) { - LOG.warn("The ordering field is not specified. EVENT_TIME_ORDERING " + log.warn("The ordering field is not specified. EVENT_TIME_ORDERING " + "merge mode requires ordering field to be set for getting the " + "event time. Using commit time-based ordering now."); } @@ -1324,7 +1322,7 @@ public class HoodieTableConfig extends HoodieConfig { setValue(TABLE_METADATA_PARTITIONS, partitions.stream().sorted().collect(Collectors.joining(CONFIG_VALUES_DELIMITER))); setValue(TABLE_METADATA_PARTITIONS_INFLIGHT, partitionsInflight.stream().sorted().collect(Collectors.joining(CONFIG_VALUES_DELIMITER))); update(metaClient.getStorage(), metaClient.getMetaPath(), getProps()); - LOG.info("MDT {} partition {} has been {}", metaClient.getBasePath(), partitionPath, enabled ? "enabled" : "disabled"); + log.info("MDT {} partition {} has been {}", metaClient.getBasePath(), partitionPath, enabled ? "enabled" : "disabled"); } /** @@ -1342,7 +1340,7 @@ public class HoodieTableConfig extends HoodieConfig { setValue(TABLE_METADATA_PARTITIONS_INFLIGHT, partitionsInflight.stream().sorted().collect(Collectors.joining(CONFIG_VALUES_DELIMITER))); update(metaClient.getStorage(), metaClient.getMetaPath(), getProps()); - LOG.info("MDT {} partitions {} have been set to inflight", metaClient.getBasePath(), partitionPaths); + log.info("MDT {} partitions {} have been set to inflight", metaClient.getBasePath(), partitionPaths); } public void setMetadataPartitionsInflight(HoodieTableMetaClient metaClient, MetadataPartitionType... partitionTypes) { diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java index e20d9d3a1220..f248dd979db8 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java @@ -74,8 +74,11 @@ import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathFilter; import org.apache.hudi.storage.StoragePathInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.io.Serializable; @@ -118,10 +121,12 @@ import static org.apache.hudi.metadata.HoodieIndexVersion.isValidIndexDefinition * @see HoodieTimeline * @since 0.3.0 */ +@NoArgsConstructor +@Getter +@Slf4j public class HoodieTableMetaClient implements Serializable { private static final long serialVersionUID = 1L; - private static final Logger LOG = LoggerFactory.getLogger(HoodieTableMetaClient.class); public static final String METADATA_STR = "metadata"; public static final String METAFOLDER_NAME = ".hoodie"; public static final String TIMELINEFOLDER_NAME = "timeline"; @@ -163,20 +168,27 @@ public class HoodieTableMetaClient implements Serializable { protected StoragePath basePath; protected StoragePath metaPath; + @Getter(AccessLevel.NONE) + @Setter private transient HoodieStorage storage; + @Getter(AccessLevel.NONE) private boolean loadActiveTimelineOnLoad; protected StorageConfiguration<?> storageConf; private HoodieTableType tableType; private TimelineLayoutVersion timelineLayoutVersion; private TimelineLayout timelineLayout; private StoragePath timelinePath; + @Getter(AccessLevel.NONE) private StoragePath timelineHistoryPath; protected HoodieTableConfig tableConfig; + @Getter(AccessLevel.NONE) protected HoodieActiveTimeline activeTimeline; private ConsistencyGuardConfig consistencyGuardConfig = ConsistencyGuardConfig.newBuilder().build(); private FileSystemRetryConfig fileSystemRetryConfig = FileSystemRetryConfig.newBuilder().build(); + @Getter(AccessLevel.NONE) protected HoodieMetaserverConfig metaserverConfig; private HoodieTimeGeneratorConfig timeGeneratorConfig; + @Getter(AccessLevel.NONE) private Option<HoodieIndexMetadata> indexMetadataOpt; private HoodieTableFormat tableFormat; @@ -187,7 +199,7 @@ public class HoodieTableMetaClient implements Serializable { protected HoodieTableMetaClient(HoodieStorage storage, String basePath, boolean loadActiveTimelineOnLoad, ConsistencyGuardConfig consistencyGuardConfig, Option<TimelineLayoutVersion> layoutVersion, HoodieTimeGeneratorConfig timeGeneratorConfig, FileSystemRetryConfig fileSystemRetryConfig) { - LOG.debug("Loading HoodieTableMetaClient from " + basePath); + log.debug("Loading HoodieTableMetaClient from {}", basePath); this.timeGeneratorConfig = timeGeneratorConfig; this.consistencyGuardConfig = consistencyGuardConfig; this.fileSystemRetryConfig = fileSystemRetryConfig; @@ -213,21 +225,13 @@ public class HoodieTableMetaClient implements Serializable { this.timelinePath = timelineLayout.getTimelinePathProvider().getTimelinePath(tableConfig, this.basePath); this.timelineHistoryPath = timelineLayout.getTimelinePathProvider().getTimelineHistoryPath(tableConfig, this.basePath); this.loadActiveTimelineOnLoad = loadActiveTimelineOnLoad; - LOG.debug("Finished Loading Table of type " + tableType + "(version=" + timelineLayoutVersion + ") from " + basePath); + log.debug("Finished Loading Table of type {}(version={}) from {}", tableType, timelineLayoutVersion, basePath); if (loadActiveTimelineOnLoad) { - LOG.info("Loading Active commit timeline for " + basePath); + log.info("Loading Active commit timeline for {}", basePath); getActiveTimeline(); } } - /** - * For serializing and de-serializing. - * - * @deprecated - */ - public HoodieTableMetaClient() { - } - public String getIndexDefinitionPath() { return tableConfig.getRelativeIndexDefinitionPath() .map(definitionPath -> new StoragePath(basePath, definitionPath).toString()) @@ -250,7 +254,7 @@ public class HoodieTableMetaClient implements Serializable { Option<HoodieIndexDefinition> existingIndexOpt = indexMetadataOpt.get().getIndex(indexName); if (existingIndexOpt.isPresent()) { if (!existingIndexOpt.get().getSourceFields().equals(indexDefinition.getSourceFields())) { - LOG.info("List of columns to index is changing. Old value {}. New value {}", existingIndexOpt.get().getSourceFields(), + log.info("List of columns to index is changing. Old value {}. New value {}", existingIndexOpt.get().getSourceFields(), indexDefinition.getSourceFields()); indexMetadataOpt.get().getIndexDefinitions().put(indexName, indexDefinition); } else { @@ -386,35 +390,6 @@ public class HoodieTableMetaClient implements Serializable { out.defaultWriteObject(); } - /** - * Returns base path of the table - */ - public StoragePath getBasePath() { - return basePath; // this invocation is cached - } - - /** - * @return Hoodie Table Type - */ - public HoodieTableType getTableType() { - return tableType; - } - - /** - * @return Meta path - */ - public StoragePath getMetaPath() { - return metaPath; - } - - public StoragePath getTimelinePath() { - return timelinePath; - } - - public HoodieTableFormat getTableFormat() { - return tableFormat; - } - /** * @return schema folder path */ @@ -492,21 +467,6 @@ public class HoodieTableMetaClient implements Serializable { return timelineHistoryPath; } - /** - * @return Table Config - */ - public HoodieTableConfig getTableConfig() { - return tableConfig; - } - - public TimelineLayoutVersion getTimelineLayoutVersion() { - return timelineLayoutVersion; - } - - public TimelineLayout getTimelineLayout() { - return timelineLayout; - } - public boolean isMetadataTable() { return HoodieTableMetadata.isMetadataTable(getBasePath()); } @@ -540,18 +500,10 @@ public class HoodieTableMetaClient implements Serializable { consistencyGuard); } - public void setStorage(HoodieStorage storage) { - this.storage = storage; - } - public HoodieStorage getRawStorage() { return getStorage().getRawStorage(); } - public StorageConfiguration<?> getStorageConf() { - return storageConf; - } - /** * Get the active instants as a timeline. * @@ -621,18 +573,6 @@ public class HoodieTableMetaClient implements Serializable { return TimelineUtils.generateInstantTime(shouldLock, timeGenerator); } - public HoodieTimeGeneratorConfig getTimeGeneratorConfig() { - return timeGeneratorConfig; - } - - public ConsistencyGuardConfig getConsistencyGuardConfig() { - return consistencyGuardConfig; - } - - public FileSystemRetryConfig getFileSystemRetryConfig() { - return fileSystemRetryConfig; - } - /** * Get the archived commits as a timeline. This is costly operation, as all data from the archived files are read. * This should not be used, unless for historical debugging purposes. @@ -696,7 +636,7 @@ public class HoodieTableMetaClient implements Serializable { Properties props, Integer timelineLayout, boolean shouldCreateTableConfig) throws IOException { - LOG.info("Initializing {} as hoodie table", basePath); + log.info("Initializing {} as hoodie table", basePath); final HoodieStorage storage = HoodieStorageUtils.getStorage(basePath, storageConf); if (!storage.exists(basePath)) { storage.createDirectory(basePath); @@ -871,31 +811,6 @@ public class HoodieTableMetaClient implements Serializable { return instantStream.sorted().collect(Collectors.toList()); } - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - HoodieTableMetaClient that = (HoodieTableMetaClient) o; - return Objects.equals(basePath, that.basePath) && tableType == that.tableType; - } - - @Override - public int hashCode() { - return Objects.hash(basePath, tableType); - } - - @Override - public String toString() { - return "HoodieTableMetaClient{" + "basePath='" + basePath + '\'' - + ", metaPath='" + metaPath + '\'' - + ", tableType=" + tableType - + '}'; - } - public void initializeBootstrapDirsIfNotExists() throws IOException { initializeBootstrapDirsIfNotExists(basePath, getStorage()); } @@ -1037,6 +952,34 @@ public class HoodieTableMetaClient implements Serializable { return getTimelineLayout().getCommitMetadataSerDe(); } + // Not using Lombok @EqualsAndHashCode/@ToString here: this class is subclassed (e.g. HoodieTableMetaserverClient), + // and we rely on the runtime subtype - exact-class matching in equals() and the declaring-class behavior below. + // Lombok would switch equals() to instanceof, making a base instance compare equal to a subtype instance. + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + HoodieTableMetaClient that = (HoodieTableMetaClient) o; + return Objects.equals(basePath, that.basePath) && tableType == that.tableType; + } + + @Override + public int hashCode() { + return Objects.hash(basePath, tableType); + } + + @Override + public String toString() { + return "HoodieTableMetaClient{" + "basePath='" + basePath + '\'' + + ", metaPath='" + metaPath + '\'' + + ", tableType=" + tableType + + '}'; + } + public static TableBuilder newTableBuilder() { return new TableBuilder(); } @@ -1044,6 +987,7 @@ public class HoodieTableMetaClient implements Serializable { /** * Builder for {@link Properties}. */ + @NoArgsConstructor(access = AccessLevel.PACKAGE) public static class TableBuilder { private HoodieTableType tableType; @@ -1093,9 +1037,6 @@ public class HoodieTableMetaClient implements Serializable { */ private final Properties others = new Properties(); - TableBuilder() { - } - public TableBuilder setTableType(HoodieTableType tableType) { this.tableType = tableType; return this; @@ -1667,7 +1608,7 @@ public class HoodieTableMetaClient implements Serializable { HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(storageConf).setBasePath(basePath) .setMetaserverConfig(props) .build(); - LOG.info("Finished initializing Table of type {} from {}", metaClient.getTableConfig().getTableType(), basePath); + log.info("Finished initializing Table of type {} from {}", metaClient.getTableConfig().getTableType(), basePath); return metaClient; } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableVersion.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableVersion.java index 945ed2366607..7ec8445809c9 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableVersion.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableVersion.java @@ -22,6 +22,11 @@ import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion; import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.exception.HoodieException; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.experimental.Accessors; + import java.util.Arrays; import java.util.List; @@ -29,7 +34,10 @@ import java.util.List; * Table's version that controls what version of writer/readers can actually read/write * to a given table. */ +@AllArgsConstructor +@Getter public enum HoodieTableVersion { + // < 0.6.0 versions ZERO(0, CollectionUtils.createImmutableList("0.3.0"), TimelineLayoutVersion.LAYOUT_VERSION_0), // 0.6.0 onwards @@ -51,26 +59,14 @@ public enum HoodieTableVersion { // 1.1 NINE(9, CollectionUtils.createImmutableList("1.1.0"), TimelineLayoutVersion.LAYOUT_VERSION_2); + @Accessors(fluent = true) // Required so that #versionCode() is generated instead of #getVersionCode() by Lombok private final int versionCode; + @Getter(AccessLevel.NONE) private final List<String> releaseVersions; private final TimelineLayoutVersion timelineLayoutVersion; - HoodieTableVersion(int versionCode, List<String> releaseVersions, TimelineLayoutVersion timelineLayoutVersion) { - this.versionCode = versionCode; - this.releaseVersions = releaseVersions; - this.timelineLayoutVersion = timelineLayoutVersion; - } - - public TimelineLayoutVersion getTimelineLayoutVersion() { - return timelineLayoutVersion; - } - - public int versionCode() { - return versionCode; - } - public static HoodieTableVersion current() { return NINE; } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java b/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java index ee2cfbf0d362..157f9d4c79ca 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java @@ -49,8 +49,7 @@ import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.util.Lazy; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import javax.annotation.concurrent.ThreadSafe; @@ -67,11 +66,10 @@ import java.util.stream.Stream; /** * Helper class to read schema from data files and log files and to convert it between different formats. */ +@Slf4j @ThreadSafe public class TableSchemaResolver { - private static final Logger LOG = LoggerFactory.getLogger(TableSchemaResolver.class); - protected final HoodieTableMetaClient metaClient; /** @@ -254,11 +252,11 @@ public class TableSchemaResolver { .map(writeStat -> new StoragePath(metaClient.getBasePath(), writeStat.getPath())); return Option.of(fetchSchemaFromFiles(filePaths)); } else { - LOG.debug("Could not find any data file written for commit, so could not get schema for table {}", metaClient.getBasePath()); + log.debug("Could not find any data file written for commit, so could not get schema for table {}", metaClient.getBasePath()); return Option.empty(); } default: - LOG.error("Unknown table type {}", metaClient.getTableType()); + log.error("Unknown table type {}", metaClient.getTableType()); throw new InvalidTableException(metaClient.getBasePath().toString()); } } @@ -373,7 +371,7 @@ public class TableSchemaResolver { HoodieSchema tableSchema = getTableSchemaFromDataFile(); return tableSchema.getField(HoodieRecord.OPERATION_METADATA_FIELD).isPresent(); } catch (Exception e) { - LOG.info("Failed to read operation field from schema ({})", e.getMessage()); + log.info("Failed to read operation field from schema ({})", e.getMessage()); return false; } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java index 612837c4dc4d..beb9b56bc66a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java @@ -219,9 +219,9 @@ public class HoodieCDCExtractor { try { Set<String> requiredActions = new HashSet<>(Arrays.asList(COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION)); HoodieActiveTimeline activeTimeLine = metaClient.getActiveTimeline(); - if (instantRange.getStartInstant().isPresent() && !metaClient.getArchivedTimeline().empty() - && InstantComparison.compareTimestamps(metaClient.getArchivedTimeline().lastInstant().get().requestedTime(), InstantComparison.GREATER_THAN, instantRange.getStartInstant().get())) { - throw new HoodieException("Start instant time " + instantRange.getStartInstant().get() + if (instantRange.getStartInstantOpt().isPresent() && !metaClient.getArchivedTimeline().empty() + && InstantComparison.compareTimestamps(metaClient.getArchivedTimeline().lastInstant().get().requestedTime(), InstantComparison.GREATER_THAN, instantRange.getStartInstantOpt().get())) { + throw new HoodieException("Start instant time " + instantRange.getStartInstantOpt().get() + " for CDC query has to be in the active timeline. Beginning of active timeline " + activeTimeLine.firstInstant().get().requestedTime()); } this.commits = activeTimeLine.getInstantsAsStream() diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java index c33df8414bd2..24fb94356a0d 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCFileSplit.java @@ -22,6 +22,8 @@ import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.FileSlice; import org.apache.hudi.common.util.Option; +import lombok.Getter; + import java.io.Serializable; import java.util.Collection; import java.util.Collections; @@ -45,7 +47,9 @@ import java.util.stream.Collectors; * For `cdcInferCase` = {@link HoodieCDCInferenceCase#REPLACE_COMMIT}, `cdcFile` is null, * `beforeFileSlice` is the current version of the file slice. */ +@Getter public class HoodieCDCFileSplit implements Serializable, Comparable<HoodieCDCFileSplit> { + /** * The instant time at which the changes happened. */ @@ -103,26 +107,6 @@ public class HoodieCDCFileSplit implements Serializable, Comparable<HoodieCDCFil this.afterFileSlice = afterFileSlice; } - public String getInstant() { - return this.instant; - } - - public HoodieCDCInferenceCase getCdcInferCase() { - return this.cdcInferCase; - } - - public List<String> getCdcFiles() { - return this.cdcFiles; - } - - public Option<FileSlice> getBeforeFileSlice() { - return this.beforeFileSlice; - } - - public Option<FileSlice> getAfterFileSlice() { - return this.afterFileSlice; - } - @Override public int compareTo(HoodieCDCFileSplit o) { int cmpResult = this.instant.compareTo(o.instant); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCOperation.java b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCOperation.java index 90540bc05a69..2cb0f19687e0 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCOperation.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCOperation.java @@ -20,9 +20,15 @@ package org.apache.hudi.common.table.cdc; import org.apache.hudi.exception.HoodieNotSupportedException; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; + /** * Enumeration of change log operation. */ +@AllArgsConstructor(access = AccessLevel.PACKAGE) +@Getter public enum HoodieCDCOperation { INSERT("i"), UPDATE("u"), @@ -30,14 +36,6 @@ public enum HoodieCDCOperation { private final String value; - HoodieCDCOperation(String value) { - this.value = value; - } - - public String getValue() { - return this.value; - } - public static HoodieCDCOperation parse(String value) { switch (value) { case "i": diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/checkpoint/Checkpoint.java b/hudi-common/src/main/java/org/apache/hudi/common/table/checkpoint/Checkpoint.java index 67248ba4adcd..eea1c32dd373 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/checkpoint/Checkpoint.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/checkpoint/Checkpoint.java @@ -19,6 +19,9 @@ package org.apache.hudi.common.table.checkpoint; +import lombok.AccessLevel; +import lombok.Getter; + import java.io.Serializable; import java.util.HashMap; import java.util.Map; @@ -27,13 +30,16 @@ import java.util.Objects; /** * Class for representing checkpoint */ +@Getter public abstract class Checkpoint implements Serializable { + public static final String CHECKPOINT_IGNORE_KEY = "deltastreamer.checkpoint.ignore_key"; protected String checkpointKey; protected String checkpointResetKey; protected String checkpointIgnoreKey; // These are extra props to be written to the commit metadata + @Getter(AccessLevel.NONE) protected Map<String, String> extraProps = new HashMap<>(); public Checkpoint setCheckpointKey(String newKey) { @@ -41,21 +47,12 @@ public abstract class Checkpoint implements Serializable { return this; } - public String getCheckpointKey() { - return checkpointKey; - } - - public String getCheckpointResetKey() { - return checkpointResetKey; - } - - public String getCheckpointIgnoreKey() { - return checkpointIgnoreKey; - } - public abstract Map<String, String> getCheckpointCommitMetadata(String overrideResetKey, String overrideIgnoreKey); + // Not using Lombok @EqualsAndHashCode/@ToString here: this class is subclassed, and we rely on + // the runtime subtype - exact-class matching in equals() and getClass().getSimpleName() in toString(). + // Lombok would bake in the declaring class (Checkpoint) and switch equals() to instanceof. @Override public int hashCode() { return Objects.hashCode(checkpointKey); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java index b1eb9e0e2c98..49c17eda75c8 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java @@ -48,8 +48,10 @@ import org.apache.hudi.internal.schema.convert.InternalSchemaConverter; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.util.ArrayDeque; @@ -87,10 +89,9 @@ import static org.apache.hudi.common.util.ValidationUtils.checkState; * <p> * This results in two I/O passes over the log file. */ +@Slf4j public abstract class AbstractHoodieLogRecordScanner { - private static final Logger LOG = LoggerFactory.getLogger(AbstractHoodieLogRecordScanner.class); - // Reader schema for the records protected final HoodieSchema readerSchema; // Latest valid instant time @@ -98,14 +99,17 @@ public abstract class AbstractHoodieLogRecordScanner { private final String latestInstantTime; protected final HoodieTableMetaClient hoodieTableMetaClient; // Merge strategy to use when combining records from log + @Getter(AccessLevel.PROTECTED) private final String payloadClassFQN; // Record's key/partition-path fields private final String recordKeyField; private final Option<String> partitionPathFieldOpt; // Partition name override + @Getter private final Option<String> partitionNameOverrideOpt; // Stateless component for merging records protected final HoodieRecordMerger recordMerger; + @Getter(AccessLevel.PROTECTED) private final TypedProperties payloadProps; // Log File Paths protected final List<String> logFilePaths; @@ -117,6 +121,7 @@ public abstract class AbstractHoodieLogRecordScanner { // optional instant range for incremental block filtering private final Option<InstantRange> instantRange; // Read the operation metadata field from the avro record + @Getter private final boolean withOperationField; private final HoodieStorage storage; // Total log files read - for metrics @@ -132,16 +137,19 @@ public abstract class AbstractHoodieLogRecordScanner { // Total number of corrupt blocks written across all log files private AtomicLong totalCorruptBlocks = new AtomicLong(0); // Store the last instant log blocks (needed to implement rollback) + @Getter private Deque<HoodieLogBlock> currentInstantLogBlocks = new ArrayDeque<>(); // Enables full scan of log records protected final boolean forceFullScan; // Progress + @Getter private float progress = 0.0f; // Populate meta fields for the records private final boolean populateMetaFields; // Record type read from log block protected final HoodieRecordType recordType; // Collect all the block instants after scanning all the log files. + @Getter private final List<String> validBlockInstants = new ArrayList<>(); // table version for compatibility private final HoodieTableVersion tableVersion; @@ -301,7 +309,7 @@ public abstract class AbstractHoodieLogRecordScanner { */ while (logFormatReaderWrapper.hasNext()) { HoodieLogFile logFile = logFormatReaderWrapper.getLogFile(); - LOG.info("Scanning log file {}", logFile); + log.info("Scanning log file {}", logFile); scannedLogFiles.add(logFile); totalLogFiles.set(scannedLogFiles.size()); // Use the HoodieLogFileReader to iterate through the blocks in the log file @@ -310,7 +318,7 @@ public abstract class AbstractHoodieLogRecordScanner { totalLogBlocks.incrementAndGet(); // Ignore the corrupt blocks. No further handling is required for them. if (logBlock.getBlockType().equals(CORRUPT_BLOCK)) { - LOG.info("Found a corrupt block in {}", logFile.getPath()); + log.info("Found a corrupt block in {}", logFile.getPath()); totalCorruptBlocks.incrementAndGet(); continue; } @@ -347,7 +355,7 @@ public abstract class AbstractHoodieLogRecordScanner { instantToBlocksMap.put(instantTime, logBlocksList); break; case COMMAND_BLOCK: - LOG.info("Reading a command block from file {}", logFile.getPath()); + log.info("Reading a command block from file {}", logFile.getPath()); // This is a command block - take appropriate action based on the command HoodieCommandBlock commandBlock = (HoodieCommandBlock) logBlock; @@ -367,8 +375,8 @@ public abstract class AbstractHoodieLogRecordScanner { } } - if (LOG.isDebugEnabled()) { - LOG.debug("Ordered instant times seen {}", orderedInstantsList); + if (log.isDebugEnabled()) { + log.debug("Ordered instant times seen {}", orderedInstantsList); } int numBlocksRolledBack = 0; @@ -424,24 +432,24 @@ public abstract class AbstractHoodieLogRecordScanner { validBlockInstants.add(compactedFinalInstantTime); } } - LOG.info("Number of applied rollback blocks {}", numBlocksRolledBack); + log.info("Number of applied rollback blocks {}", numBlocksRolledBack); - if (LOG.isDebugEnabled()) { - LOG.info("Final view of the Block time to compactionBlockMap {}", blockTimeToCompactionBlockTimeMap); + if (log.isDebugEnabled()) { + log.info("Final view of the Block time to compactionBlockMap {}", blockTimeToCompactionBlockTimeMap); } // merge the last read block when all the blocks are done reading if (!currentInstantLogBlocks.isEmpty() && !skipProcessingBlocks) { - LOG.info("Merging the final data blocks"); + log.info("Merging the final data blocks"); processQueuedBlocksForInstant(currentInstantLogBlocks, scannedLogFiles.size(), keySpecOpt); } // Done progress = 1.0f; } catch (IOException e) { - LOG.error("Got IOException when reading log file", e); + log.error("Got IOException when reading log file", e); throw new HoodieIOException("IOException when reading log file ", e); } catch (Exception e) { - LOG.error("Got exception when reading log file", e); + log.error("Got exception when reading log file", e); throw new HoodieException("Exception when reading log file ", e); } finally { try { @@ -450,7 +458,7 @@ public abstract class AbstractHoodieLogRecordScanner { } } catch (IOException ioe) { // Eat exception as we do not want to mask the original exception that can happen - LOG.error("Unable to close log format reader", ioe); + log.error("Unable to close log format reader", ioe); } } } @@ -506,7 +514,7 @@ public abstract class AbstractHoodieLogRecordScanner { private void processQueuedBlocksForInstant(Deque<HoodieLogBlock> logBlocks, int numLogFilesSeen, Option<KeySpec> keySpecOpt) throws Exception { while (!logBlocks.isEmpty()) { - LOG.info("Number of remaining logblocks to merge {}", logBlocks.size()); + log.info("Number of remaining logblocks to merge {}", logBlocks.size()); // poll the element at the bottom of the stack since that's the order it was inserted HoodieLogBlock lastBlock = logBlocks.pollLast(); switch (lastBlock.getBlockType()) { @@ -519,7 +527,7 @@ public abstract class AbstractHoodieLogRecordScanner { Arrays.stream(((HoodieDeleteBlock) lastBlock).getRecordsToDelete()).forEach(this::processNextDeletedRecord); break; case CORRUPT_BLOCK: - LOG.warn("Found a corrupt block which was not rolled back"); + log.warn("Found a corrupt block which was not rolled back"); break; default: break; @@ -535,13 +543,6 @@ public abstract class AbstractHoodieLogRecordScanner { return !forceFullScan; } - /** - * Return progress of scanning as a float between 0.0 to 1.0. - */ - public float getProgress() { - return progress; - } - public long getTotalLogFiles() { return totalLogFiles.get(); } @@ -554,14 +555,6 @@ public abstract class AbstractHoodieLogRecordScanner { return totalLogBlocks.get(); } - protected String getPayloadClassFQN() { - return payloadClassFQN; - } - - public Option<String> getPartitionNameOverride() { - return partitionNameOverrideOpt; - } - public long getTotalRollbacks() { return totalRollbacks.get(); } @@ -570,14 +563,6 @@ public abstract class AbstractHoodieLogRecordScanner { return totalCorruptBlocks.get(); } - public boolean isWithOperationField() { - return withOperationField; - } - - protected TypedProperties getPayloadProps() { - return payloadProps; - } - /** * Key specification with a list of column names. */ @@ -595,16 +580,11 @@ public abstract class AbstractHoodieLogRecordScanner { } } + @AllArgsConstructor + @Getter private static class FullKeySpec implements KeySpec { - private final List<String> keys; - private FullKeySpec(List<String> keys) { - this.keys = keys; - } - @Override - public List<String> getKeys() { - return keys; - } + private final List<String> keys; @Override public boolean isFullKey() { @@ -612,12 +592,10 @@ public abstract class AbstractHoodieLogRecordScanner { } } + @AllArgsConstructor private static class PrefixKeySpec implements KeySpec { - private final List<String> keysPrefixes; - private PrefixKeySpec(List<String> keysPrefixes) { - this.keysPrefixes = keysPrefixes; - } + private final List<String> keysPrefixes; @Override public List<String> getKeys() { @@ -630,14 +608,6 @@ public abstract class AbstractHoodieLogRecordScanner { } } - public Deque<HoodieLogBlock> getCurrentInstantLogBlocks() { - return currentInstantLogBlocks; - } - - public List<String> getValidBlockInstants() { - return validBlockInstants; - } - private Pair<ClosableIterator<HoodieRecord>, HoodieSchema> getRecordsIterator( HoodieDataBlock dataBlock, Option<KeySpec> keySpecOpt) throws IOException { ClosableIterator<HoodieRecord> blockRecordsIterator; diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java index 332a9c920c6e..159662f73bb9 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java @@ -43,8 +43,9 @@ import org.apache.hudi.internal.schema.InternalSchema; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.util.ArrayDeque; @@ -74,9 +75,9 @@ import static org.apache.hudi.common.util.ValidationUtils.checkState; * * @param <T> type of engine-specific record representation. */ +@Slf4j public abstract class BaseHoodieLogRecordReader<T> { - private static final Logger LOG = LoggerFactory.getLogger(BaseHoodieLogRecordReader.class); public static final String LOG_BLOCK_FULL_READ_DURATION_IN_MILLIS = "logBlockFullReadDurationInMillis"; public static final String BLOCK_SIZE_IN_BYTES = "blockSizeInBytes"; public static final String TOTAL_RECORDS_PRESENT_IN_LOG_BLOCK = "totalRecordsPresentInLogBlock"; @@ -89,13 +90,16 @@ public abstract class BaseHoodieLogRecordReader<T> { protected final HoodieReaderContext<T> readerContext; protected final HoodieTableMetaClient hoodieTableMetaClient; // Merge strategy to use when combining records from log + @Getter(AccessLevel.PROTECTED) private final String payloadClassFQN; // Record's key/partition-path fields private final String recordKeyField; // Partition name override + @Getter private final Option<String> partitionNameOverrideOpt; // Ordering fields protected final String orderingFields; + @Getter(AccessLevel.PROTECTED) private final TypedProperties payloadProps; // Log File Paths protected final List<HoodieLogFile> logFiles; @@ -107,6 +111,7 @@ public abstract class BaseHoodieLogRecordReader<T> { // optional instant range for incremental block filtering private final Option<InstantRange> instantRange; // Read the operation metadata field from the avro record + @Getter private final boolean withOperationField; // FileSystem private final HoodieStorage storage; @@ -129,15 +134,18 @@ public abstract class BaseHoodieLogRecordReader<T> { // Scan duration in milliseconds private AtomicLong blocksScanDuration = new AtomicLong(0); // Store the last instant log blocks (needed to implement rollback) + @Getter private Deque<HoodieLogBlock> currentInstantLogBlocks = new ArrayDeque<>(); // Enables full scan of log records protected final boolean forceFullScan; // Progress + @Getter private float progress = 0.0f; - // Record type read from log block // Collect all the block instants after scanning all the log files. + @Getter private final List<String> validBlockInstants = new ArrayList<>(); // Block-level scan stats for processed data blocks. + @Getter private List<Map<String, Object>> blocksStats = new ArrayList<>(); protected HoodieFileGroupRecordBuffer<T> recordBuffer; // Allows to consider inflight instants while merging log records @@ -271,7 +279,7 @@ public abstract class BaseHoodieLogRecordReader<T> { */ while (logFormatReaderWrapper.hasNext()) { HoodieLogFile logFile = logFormatReaderWrapper.getLogFile(); - LOG.debug("Scanning log file {}", logFile); + log.debug("Scanning log file {}", logFile); scannedLogFiles.add(logFile); totalLogFiles.set(scannedLogFiles.size()); // Use the HoodieLogFileReader to iterate through the blocks in the log file @@ -279,11 +287,11 @@ public abstract class BaseHoodieLogRecordReader<T> { logBlock.getBlockContentLocation() .map(contentLocation -> totalLogBlocksSize.addAndGet(contentLocation.getBlockSize())); final String instantTime = logBlock.getLogBlockHeader().get(INSTANT_TIME); - LOG.debug("Scanning log block with instant time {}", instantTime); + log.debug("Scanning log block with instant time {}", instantTime); totalLogBlocks.incrementAndGet(); // Ignore the corrupt blocks. No further handling is required for them. if (logBlock.getBlockType().equals(CORRUPT_BLOCK)) { - LOG.debug("Found a corrupt block in {}", logFile.getPath()); + log.debug("Found a corrupt block in {}", logFile.getPath()); totalCorruptBlocks.incrementAndGet(); continue; } @@ -314,7 +322,7 @@ public abstract class BaseHoodieLogRecordReader<T> { case AVRO_DATA_BLOCK: case PARQUET_DATA_BLOCK: case DELETE_BLOCK: - LOG.debug("Reading a {} block with instant time {}", + log.debug("Reading a {} block with instant time {}", logBlock.getBlockType() == HoodieLogBlock.HoodieLogBlockType.DELETE_BLOCK ? "delete" : "data", instantTime); List<HoodieLogBlock> logBlocksList = instantToBlocksMap.getOrDefault(instantTime, new ArrayList<>()); @@ -326,7 +334,7 @@ public abstract class BaseHoodieLogRecordReader<T> { instantToBlocksMap.put(instantTime, logBlocksList); break; case COMMAND_BLOCK: - LOG.debug("Reading a command block from file {}", logFile.getPath()); + log.debug("Reading a command block from file {}", logFile.getPath()); // This is a command block - take appropriate action based on the command HoodieCommandBlock commandBlock = (HoodieCommandBlock) logBlock; @@ -341,10 +349,10 @@ public abstract class BaseHoodieLogRecordReader<T> { if (rolledBackBlocks != null) { numBlocksRolledBack += rolledBackBlocks.size(); } - LOG.debug("Reading a rollback block with instant {} and target instant {}", + log.debug("Reading a rollback block with instant {} and target instant {}", instantTime, targetInstantForCommandBlock); } else { - LOG.error("Reading a command block with instant {} whose operation is not supported", instantTime); + log.error("Reading a command block with instant {} whose operation is not supported", instantTime); throw new UnsupportedOperationException("Command type not yet supported."); } break; @@ -353,8 +361,8 @@ public abstract class BaseHoodieLogRecordReader<T> { } } - LOG.info("Ordered instant times seen {}", orderedInstantsList); - LOG.info("Targeted instants that are rolled back are {}", targetRollbackInstants); + log.info("Ordered instant times seen {}", orderedInstantsList); + log.info("Targeted instants that are rolled back are {}", targetRollbackInstants); // All the block's instants time that are added to the queue are collected in this set. Set<String> instantTimesIncluded = new HashSet<>(); @@ -377,7 +385,7 @@ public abstract class BaseHoodieLogRecordReader<T> { // For compacted blocks COMPACTED_BLOCK_TIMES entry is present under its headers. if (firstBlock.getLogBlockHeader().containsKey(COMPACTED_BLOCK_TIMES)) { - LOG.debug("For instant time {}, compacted block instants are {}", + log.debug("For instant time {}, compacted block instants are {}", instantTime, firstBlock.getLogBlockHeader().get(COMPACTED_BLOCK_TIMES)); // When compacted blocks are seen update the blockTimeToCompactionBlockTimeMap. Arrays.stream(firstBlock.getLogBlockHeader().get(COMPACTED_BLOCK_TIMES).split(",")) @@ -410,20 +418,20 @@ public abstract class BaseHoodieLogRecordReader<T> { } } Collections.reverse(validBlockInstants); - LOG.debug("Number of applied rollback blocks {}", numBlocksRolledBack); - LOG.info("Total valid instants found are {}. Instants are {}", validBlockInstants.size(), validBlockInstants); + log.debug("Number of applied rollback blocks {}", numBlocksRolledBack); + log.info("Total valid instants found are {}. Instants are {}", validBlockInstants.size(), validBlockInstants); if (ignoredBlockCount > 0) { - LOG.info("Ignored {} log blocks from {} instants not in the range: {}", ignoredBlockCount, ignoredInstants.size(), ignoredInstants); + log.info("Ignored {} log blocks from {} instants not in the range: {}", ignoredBlockCount, ignoredInstants.size(), ignoredInstants); } - if (LOG.isDebugEnabled()) { - LOG.debug("Final view of the Block time to compactionBlockMap {}", blockTimeToCompactionBlockTimeMap); + if (log.isDebugEnabled()) { + log.debug("Final view of the Block time to compactionBlockMap {}", blockTimeToCompactionBlockTimeMap); } totalValidLogBlocks.set(currentInstantLogBlocks.size()); blocksScanDuration.set(scanTimer.endTimer()); // merge the last read block when all the blocks are done reading if (!currentInstantLogBlocks.isEmpty() && !skipProcessingBlocks) { - LOG.debug("Merging the final data blocks"); + log.debug("Merging the final data blocks"); processQueuedBlocksForInstant(currentInstantLogBlocks, scannedLogFiles.size(), keySpecOpt); } // Done @@ -432,10 +440,10 @@ public abstract class BaseHoodieLogRecordReader<T> { totalLogRecords.set(recordBuffer.getTotalLogRecords()); } } catch (IOException e) { - LOG.error("Got IOException when reading log file", e); + log.error("Got IOException when reading log file", e); throw new HoodieIOException("IOException when reading log file ", e); } catch (Exception e) { - LOG.error("Got exception when reading log file", e); + log.error("Got exception when reading log file", e); throw new HoodieException("Exception when reading log file ", e); } finally { try { @@ -444,18 +452,18 @@ public abstract class BaseHoodieLogRecordReader<T> { } } catch (IOException ioe) { // Eat exception as we do not want to mask the original exception that can happen - LOG.error("Unable to close log format reader", ioe); + log.error("Unable to close log format reader", ioe); } if (!logFiles.isEmpty()) { try { StoragePath path = logFiles.get(0).getPath(); - LOG.info("Finished scanning log files. FileId: {}, LogFileInstantTime: {}, " + log.info("Finished scanning log files. FileId: {}, LogFileInstantTime: {}, " + "Total log files: {}, Total log blocks: {}, Total rollbacks: {}, Total corrupt blocks: {}", FSUtils.getFileIdFromLogPath(path), FSUtils.getDeltaCommitTimeFromLogPath(path), totalLogFiles.get(), totalLogBlocks.get(), totalRollbacks.get(), totalCorruptBlocks.get()); } catch (Exception e) { - LOG.warn("Could not extract fileId from log path", e); - LOG.info("Finished scanning log files. " + log.warn("Could not extract fileId from log path", e); + log.info("Finished scanning log files. " + "Total log files: {}, Total log blocks: {}, Total rollbacks: {}, Total corrupt blocks: {}", totalLogFiles.get(), totalLogBlocks.get(), totalRollbacks.get(), totalCorruptBlocks.get()); } @@ -469,7 +477,7 @@ public abstract class BaseHoodieLogRecordReader<T> { private void processQueuedBlocksForInstant(Deque<HoodieLogBlock> logBlocks, int numLogFilesSeen, Option<KeySpec> keySpecOpt) throws Exception { while (!logBlocks.isEmpty()) { - LOG.debug("Number of remaining logblocks to merge {}", logBlocks.size()); + log.debug("Number of remaining logblocks to merge {}", logBlocks.size()); // poll the element at the bottom of the stack since that's the order it was inserted HoodieLogBlock lastBlock = logBlocks.pollLast(); switch (lastBlock.getBlockType()) { @@ -482,7 +490,7 @@ public abstract class BaseHoodieLogRecordReader<T> { recordBuffer.processDeleteBlock((HoodieDeleteBlock) lastBlock); break; case CORRUPT_BLOCK: - LOG.warn("Found a corrupt block which was not rolled back"); + log.warn("Found a corrupt block which was not rolled back"); break; default: break; @@ -494,7 +502,7 @@ public abstract class BaseHoodieLogRecordReader<T> { private void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec> keySpecOpt) throws IOException { String blockInstantTime = dataBlock.getLogBlockHeader().get(INSTANT_TIME); - LOG.debug("Processing log block with instant time {}", blockInstantTime); + log.debug("Processing log block with instant time {}", blockInstantTime); long totalLogRecordsBefore = recordBuffer != null ? recordBuffer.getTotalLogRecords() : 0L; HoodieTimer blockReadTimer = HoodieTimer.start(); recordBuffer.processDataBlock(dataBlock, keySpecOpt); @@ -507,7 +515,7 @@ public abstract class BaseHoodieLogRecordReader<T> { .map(contentLocation -> blockReadMetrics.put(BLOCK_SIZE_IN_BYTES, contentLocation.getBlockSize())); blockReadMetrics.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME.toString(), blockInstantTime); blocksStats.add(blockReadMetrics); - LOG.debug("For log block, scan metrics are {}", blockReadMetrics); + log.debug("For log block, scan metrics are {}", blockReadMetrics); } private boolean shouldLookupRecords() { @@ -516,13 +524,6 @@ public abstract class BaseHoodieLogRecordReader<T> { return !forceFullScan; } - /** - * Return progress of scanning as a float between 0.0 to 1.0. - */ - public float getProgress() { - return progress; - } - public long getTotalLogFiles() { return totalLogFiles.get(); } @@ -543,22 +544,10 @@ public abstract class BaseHoodieLogRecordReader<T> { return totalValidLogBlocks.get(); } - public List<Map<String, Object>> getBlocksStats() { - return blocksStats; - } - public long getBlocksScanDuration() { return blocksScanDuration.get(); } - protected String getPayloadClassFQN() { - return payloadClassFQN; - } - - public Option<String> getPartitionNameOverride() { - return partitionNameOverrideOpt; - } - public long getTotalRollbacks() { return totalRollbacks.get(); } @@ -567,22 +556,6 @@ public abstract class BaseHoodieLogRecordReader<T> { return totalCorruptBlocks.get(); } - public boolean isWithOperationField() { - return withOperationField; - } - - protected TypedProperties getPayloadProps() { - return payloadProps; - } - - public Deque<HoodieLogBlock> getCurrentInstantLogBlocks() { - return currentInstantLogBlocks; - } - - public List<String> getValidBlockInstants() { - return validBlockInstants; - } - /** * Builder used to build {@code AbstractHoodieLogRecordScanner}. */ diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/FullKeySpec.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/FullKeySpec.java index ede7918649b4..a186abd73e9a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/FullKeySpec.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/FullKeySpec.java @@ -19,6 +19,9 @@ package org.apache.hudi.common.table.log; +import lombok.AllArgsConstructor; +import lombok.Getter; + import java.util.List; /** @@ -26,17 +29,11 @@ import java.util.List; * That is, the comparison between a record key and an element * of the set is {@link String#equals}. */ +@AllArgsConstructor +@Getter public class FullKeySpec implements KeySpec { - private final List<String> keys; - - public FullKeySpec(List<String> keys) { - this.keys = keys; - } - @Override - public List<String> getKeys() { - return keys; - } + private final List<String> keys; @Override public boolean isFullKey() { diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java index 2c1d91e7b7c1..03ae4a2c4c07 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java @@ -44,8 +44,8 @@ import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StorageSchemes; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import javax.annotation.Nullable; @@ -63,14 +63,15 @@ import static org.apache.hudi.common.util.ValidationUtils.checkState; * Scans a log file and provides block level iterator on the log file Loads the entire block contents in memory Can emit * either a DataBlock, CommandBlock, DeleteBlock or CorruptBlock (if one is found). */ +@Slf4j public class HoodieLogFileReader implements HoodieLogFormat.Reader { public static final int DEFAULT_BUFFER_SIZE = 16 * 1024 * 1024; // 16 MB - private static final int BLOCK_SCAN_READ_BUFFER_SIZE = 1024 * 1024; // 1 MB - private static final Logger LOG = LoggerFactory.getLogger(HoodieLogFileReader.class); + private static final int BLOCK_SCAN_READ_BUFFER_SIZE = 1024 * 1024; private static final String REVERSE_LOG_READER_HAS_NOT_BEEN_ENABLED = "Reverse log reader has not been enabled"; private final HoodieStorage storage; + @Getter private final HoodieLogFile logFile; private final int bufferSize; private final byte[] magicBuffer = new byte[6]; @@ -118,11 +119,6 @@ public class HoodieLogFileReader implements HoodieLogFormat.Reader { } } - @Override - public HoodieLogFile getLogFile() { - return logFile; - } - // TODO : convert content and block length to long by using ByteBuffer, raw byte [] allows // for max of Integer size private HoodieLogBlock readBlock() throws IOException { @@ -244,12 +240,12 @@ public class HoodieLogFileReader implements HoodieLogFormat.Reader { } private HoodieLogBlock createCorruptBlock(long blockStartPos) throws IOException { - LOG.info("Log {} has a corrupted block at {}", logFile, blockStartPos); + log.info("Log {} has a corrupted block at {}", logFile, blockStartPos); inputStream.seek(blockStartPos); long nextBlockOffset = scanForNextAvailableBlockOffset(); // Rewind to the initial start and read corrupted bytes till the nextBlockOffset inputStream.seek(blockStartPos); - LOG.info("Next available block in {} starts at {}", logFile, nextBlockOffset); + log.info("Next available block in {} starts at {}", logFile, nextBlockOffset); int corruptedBlockSize = (int) (nextBlockOffset - blockStartPos); long contentPosition = inputStream.getPos(); Option<byte[]> corruptedBytes = HoodieLogBlock.tryReadContent(inputStream, corruptedBlockSize, true); @@ -276,7 +272,7 @@ public class HoodieLogFileReader implements HoodieLogFormat.Reader { // So we have to shorten the footer block size by the size of magic hash blockSizeFromFooter = inputStream.readLong() - magicBuffer.length; } catch (EOFException e) { - LOG.info("Found corrupted block in file {} with block size({}) running past EOF", logFile, blocksize); + log.info("Found corrupted block in file {} with block size({}) running past EOF", logFile, blocksize); // this is corrupt // This seek is required because contract of seek() is different for naked DFSInputStream vs BufferedFSInputStream // release-3.1.0-RC1/DFSInputStream.java#L1455 @@ -286,7 +282,7 @@ public class HoodieLogFileReader implements HoodieLogFormat.Reader { } if (blocksize != blockSizeFromFooter) { - LOG.info("Found corrupted block in file {}. Header block size({}) did not match the footer block size({})", logFile, blocksize, blockSizeFromFooter); + log.info("Found corrupted block in file {}. Header block size({}) did not match the footer block size({})", logFile, blocksize, blockSizeFromFooter); inputStream.seek(currentPos); return true; } @@ -297,7 +293,7 @@ public class HoodieLogFileReader implements HoodieLogFormat.Reader { return false; } catch (CorruptedLogFileException e) { // This is a corrupted block - LOG.info("Found corrupted block in file {}. No magic hash found right after footer block size entry", logFile); + log.info("Found corrupted block in file {}. No magic hash found right after footer block size entry", logFile); return true; } finally { inputStream.seek(currentPos); @@ -330,7 +326,7 @@ public class HoodieLogFileReader implements HoodieLogFormat.Reader { @Override public void close() throws IOException { if (!closed) { - LOG.info("Closing Log file reader {}", logFile.getFileName()); + log.info("Closing Log file reader {}", logFile.getFileName()); if (null != this.inputStream) { this.inputStream.close(); } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java index a675e0d9da89..3877a682240d 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java @@ -29,6 +29,8 @@ import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; import lombok.Getter; import lombok.extern.slf4j.Slf4j; @@ -246,18 +248,12 @@ public interface HoodieLogFormat { * A set of feature flags associated with a log format. Versions are changed when the log format changes. TODO(na) - * Implement policies around major/minor versions */ + @AllArgsConstructor(access = AccessLevel.PACKAGE) + @Getter abstract class LogFormatVersion { private final int version; - LogFormatVersion(int version) { - this.version = version; - } - - public int getVersion() { - return version; - } - public abstract boolean hasMagicHeader(); public abstract boolean hasContent(); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java index 77c3e78fcc32..1c342fecf39c 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java @@ -25,8 +25,7 @@ import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.internal.schema.InternalSchema; import org.apache.hudi.storage.HoodieStorage; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.util.List; @@ -34,6 +33,7 @@ import java.util.List; /** * Hoodie log format reader. */ +@Slf4j public class HoodieLogFormatReader implements HoodieLogFormat.Reader { private final List<HoodieLogFile> logFiles; @@ -45,8 +45,6 @@ public class HoodieLogFormatReader implements HoodieLogFormat.Reader { private final boolean enableInlineReading; private final int bufferSize; - private static final Logger LOG = LoggerFactory.getLogger(HoodieLogFormatReader.class); - HoodieLogFormatReader(HoodieStorage storage, List<HoodieLogFile> logFiles, HoodieSchema readerSchema, boolean reverseLogReader, int bufferSize, boolean enableRecordLookups, String recordKeyField, InternalSchema internalSchema) throws IOException { @@ -91,7 +89,7 @@ public class HoodieLogFormatReader implements HoodieLogFormat.Reader { } catch (IOException io) { throw new HoodieIOException("unable to initialize read with log file ", io); } - LOG.debug("Moving to the next reader for logfile {}", currentReader.getLogFile()); + log.debug("Moving to the next reader for logfile {}", currentReader.getLogFile()); return hasNext(); } return false; diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordReader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordReader.java index b3f1eeaa988e..1ee8e7560d05 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordReader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordReader.java @@ -34,8 +34,8 @@ import org.apache.hudi.expression.Predicates; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import java.io.Closeable; import java.io.Serializable; @@ -52,9 +52,11 @@ import static org.apache.hudi.common.fs.FSUtils.getRelativePartitionPath; * * @param <T> type of engine-specific record representation. */ +@Getter +@Slf4j public class HoodieMergedLogRecordReader<T> extends BaseHoodieLogRecordReader<T> implements Iterable<BufferedRecord<T>>, Closeable { - private static final Logger LOG = LoggerFactory.getLogger(HoodieMergedLogRecordReader.class); + // A timer for calculating elapsed time in millis public final HoodieTimer timer = HoodieTimer.create(); // count of merged records in log @@ -102,8 +104,8 @@ public class HoodieMergedLogRecordReader<T> extends BaseHoodieLogRecordReader<T> this.totalTimeTakenToReadAndMergeBlocks = timer.endTimer(); this.numMergedRecordsInLog = recordBuffer.size(); - LOG.info("Number of log files scanned => {}", logFiles.size()); - LOG.info("Number of entries in Map => {}", recordBuffer.size()); + log.info("Number of log files scanned => {}", logFiles.size()); + log.info("Number of entries in Map => {}", recordBuffer.size()); } static Option<KeySpec> createKeySpec(Option<Predicate> filter) { @@ -134,10 +136,6 @@ public class HoodieMergedLogRecordReader<T> extends BaseHoodieLogRecordReader<T> return recordBuffer.getLogRecords(); } - public long getNumMergedRecordsInLog() { - return numMergedRecordsInLog; - } - /** * Returns the builder for {@code HoodieMergedLogRecordReader}. */ @@ -145,10 +143,6 @@ public class HoodieMergedLogRecordReader<T> extends BaseHoodieLogRecordReader<T> return new Builder<>(); } - public long getTotalTimeTakenToReadAndMergeBlocks() { - return totalTimeTakenToReadAndMergeBlocks; - } - @Override public void close() { // No op. diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java index 318533859746..f7521fc207c5 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java @@ -47,8 +47,8 @@ import org.apache.hudi.internal.schema.InternalSchema; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.storage.StoragePath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import javax.annotation.concurrent.NotThreadSafe; @@ -81,10 +81,10 @@ import static org.apache.hudi.common.util.ValidationUtils.checkArgument; * This results in two I/O passes over the log file. */ @NotThreadSafe +@Slf4j public class HoodieMergedLogRecordScanner extends AbstractHoodieLogRecordScanner implements Iterable<HoodieRecord>, Closeable { - private static final Logger LOG = LoggerFactory.getLogger(HoodieMergedLogRecordScanner.class); // A timer for calculating elapsed time in millis public final HoodieTimer timer = HoodieTimer.create(); // Map of compacted/merged records @@ -92,9 +92,11 @@ public class HoodieMergedLogRecordScanner extends AbstractHoodieLogRecordScanner // Set of already scanned prefixes allowing us to avoid scanning same prefixes again private final Set<String> scannedPrefixes; // count of merged records in log + @Getter private long numMergedRecordsInLog; private final long maxMemorySizeInBytes; // Stores the total time taken to perform reading and merging of log blocks + @Getter private long totalTimeTakenToReadAndMergeBlocks; private final String[] orderingFields; private final DeleteContext deleteContext; @@ -220,8 +222,8 @@ public class HoodieMergedLogRecordScanner extends AbstractHoodieLogRecordScanner this.totalTimeTakenToReadAndMergeBlocks = timer.endTimer(); this.numMergedRecordsInLog = records.size(); - if (LOG.isInfoEnabled()) { - LOG.info("Scanned {} log files with stats: MaxMemoryInBytes => {}, MemoryBasedMap => {} entries, {} total bytes, DiskBasedMap => {} entries, {} total bytes", + if (log.isInfoEnabled()) { + log.info("Scanned {} log files with stats: MaxMemoryInBytes => {}, MemoryBasedMap => {} entries, {} total bytes, DiskBasedMap => {} entries, {} total bytes", logFilePaths.size(), maxMemorySizeInBytes, records.getInMemoryMapNumEntries(), records.getCurrentInMemoryMapSize(), records.getDiskBasedMapNumEntries(), records.getSizeOfFileOnDiskInBytes()); } @@ -240,10 +242,6 @@ public class HoodieMergedLogRecordScanner extends AbstractHoodieLogRecordScanner return recordMerger.getRecordType(); } - public long getNumMergedRecordsInLog() { - return numMergedRecordsInLog; - } - /** * Returns the builder for {@code HoodieMergedLogRecordScanner}. */ @@ -313,10 +311,6 @@ public class HoodieMergedLogRecordScanner extends AbstractHoodieLogRecordScanner } } - public long getTotalTimeTakenToReadAndMergeBlocks() { - return totalTimeTakenToReadAndMergeBlocks; - } - @Override public void close() { if (records != null) { diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java index 9dd56cc66182..eda285d478d8 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java @@ -21,6 +21,10 @@ package org.apache.hudi.common.table.log; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.ValidationUtils; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.NoArgsConstructor; + import java.io.Serializable; import java.util.Arrays; import java.util.Collections; @@ -37,6 +41,7 @@ import static org.apache.hudi.common.table.timeline.InstantComparison.compareTim /** * An instant range used for incremental reader filtering. */ +@Getter public abstract class InstantRange implements Serializable { private static final long serialVersionUID = 1L; @@ -55,14 +60,6 @@ public abstract class InstantRange implements Serializable { return new Builder(); } - public Option<String> getStartInstant() { - return startInstantOpt; - } - - public Option<String> getEndInstant() { - return endInstantOpt; - } - public abstract boolean isInRange(String instant); public abstract RangeType getRangeType(); @@ -248,6 +245,7 @@ public abstract class InstantRange implements Serializable { /** * Builder for {@link InstantRange}. */ + @NoArgsConstructor(access = AccessLevel.PRIVATE) public static class Builder { private String startInstant; private String endInstant; @@ -256,9 +254,6 @@ public abstract class InstantRange implements Serializable { private Set<String> explicitInstants; private List<InstantRange> instantRanges; - private Builder() { - } - public Builder startInstant(String startInstant) { this.startInstant = startInstant; return this; diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieCommandBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieCommandBlock.java index 965c57309b65..ea4c25db661a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieCommandBlock.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieCommandBlock.java @@ -22,6 +22,8 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.io.SeekableDataInputStream; import org.apache.hudi.storage.HoodieStorage; +import lombok.Getter; + import java.io.ByteArrayOutputStream; import java.util.HashMap; import java.util.Map; @@ -30,6 +32,7 @@ import java.util.function.Supplier; /** * Command block issues a specific command to the scanner. */ +@Getter public class HoodieCommandBlock extends HoodieLogBlock { private final HoodieCommandBlockTypeEnum type; @@ -53,10 +56,6 @@ public class HoodieCommandBlock extends HoodieLogBlock { HoodieCommandBlockTypeEnum.values()[Integer.parseInt(header.get(HeaderMetadataType.COMMAND_BLOCK_TYPE))]; } - public HoodieCommandBlockTypeEnum getType() { - return type; - } - @Override public HoodieLogBlockType getBlockType() { return HoodieLogBlockType.COMMAND_BLOCK; diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java index aee07ea947a7..7a0abf13a37a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java @@ -30,8 +30,8 @@ import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.io.SeekableDataInputStream; import org.apache.hudi.storage.HoodieStorage; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -57,8 +57,8 @@ import static org.apache.hudi.common.util.ValidationUtils.checkState; * 2. Total number of records in the block * 3. Actual serialized content of the records */ +@Slf4j public abstract class HoodieDataBlock extends HoodieLogBlock { - private static final Logger LOG = LoggerFactory.getLogger(HoodieDataBlock.class); // TODO rebase records/content to leverage Either to warrant // that they are mutex (used by read/write flows respectively) @@ -67,6 +67,7 @@ public abstract class HoodieDataBlock extends HoodieLogBlock { /** * Key field's name w/in the record's schema */ + @Getter private final String keyFieldName; private final boolean enablePointLookups; @@ -132,10 +133,6 @@ public abstract class HoodieDataBlock extends HoodieLogBlock { return serializeRecords(records.get(), storage); } - public String getKeyFieldName() { - return keyFieldName; - } - public boolean containsPartialUpdates() { return getLogBlockHeader().containsKey(HeaderMetadataType.IS_PARTIAL) && Boolean.parseBoolean(getLogBlockHeader().get(HeaderMetadataType.IS_PARTIAL)); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java index 54e7f301e6a8..0d69dab14258 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java @@ -32,6 +32,7 @@ import org.apache.hudi.io.SeekableDataInputStream; import org.apache.hudi.storage.HoodieStorage; import org.apache.hudi.util.Lazy; +import lombok.extern.slf4j.Slf4j; import org.apache.avro.generic.GenericDatumReader; import org.apache.avro.generic.GenericRecord; import org.apache.avro.io.BinaryDecoder; @@ -41,8 +42,6 @@ import org.apache.avro.io.DatumWriter; import org.apache.avro.io.DecoderFactory; import org.apache.avro.io.EncoderFactory; import org.apache.avro.specific.SpecificDatumWriter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -62,8 +61,9 @@ import static org.apache.hudi.avro.HoodieAvroWrapperUtils.wrapValueIntoAvro; /** * Delete block contains a list of keys to be deleted from scanning the blocks so far. */ +@Slf4j public class HoodieDeleteBlock extends HoodieLogBlock { - private static final Logger LOG = LoggerFactory.getLogger(HoodieDeleteBlock.class); + /** * These static builders are added to avoid performance issue in Avro 1.10. * You can find more details in HoodieAvroUtils, HUDI-3834, and AVRO-3048. diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java index 659cbf8c8d18..40b1bd5dd0a8 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java @@ -38,6 +38,7 @@ import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; import org.apache.hudi.storage.inline.InLineFSUtils; +import lombok.extern.slf4j.Slf4j; import org.apache.avro.generic.IndexedRecord; import java.io.ByteArrayOutputStream; @@ -56,6 +57,7 @@ import static org.apache.hudi.common.util.ValidationUtils.checkState; * HoodieHFileDataBlock contains a list of records stored inside an HFile format. It is used with the HFile * base file format. */ +@Slf4j public class HoodieHFileDataBlock extends HoodieDataBlock { private final Option<String> compressionCodec; // This path is used for constructing HFile reader context, which should not be diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java index 554b83b3455d..0b703ed7ac91 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java @@ -28,9 +28,11 @@ import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.io.SeekableDataInputStream; import org.apache.hudi.storage.HoodieStorage; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; import org.roaringbitmap.longlong.Roaring64NavigableMap; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -55,8 +57,11 @@ import static org.apache.hudi.common.util.ValidationUtils.checkState; /** * Abstract class defining a block in HoodieLogFile. */ +@AllArgsConstructor +@Getter +@Slf4j public abstract class HoodieLogBlock { - private static final Logger LOG = LoggerFactory.getLogger(HoodieLogBlock.class); + /** * The current version of the log block. Anytime the logBlock format changes this version needs to be bumped and * corresponding changes need to be made to {@link HoodieLogBlockVersion} TODO : Change this to a class, something @@ -65,32 +70,24 @@ public abstract class HoodieLogBlock { */ public static int version = 3; // Header for each log block + @Nonnull private final Map<HeaderMetadataType, String> logBlockHeader; // Footer for each log block + @Nonnull private final Map<FooterMetadataType, String> logBlockFooter; // Location of a log block on disk + @Nonnull private final Option<HoodieLogBlockContentLocation> blockContentLocation; // data for a specific block + @Nonnull private Option<byte[]> content; + @Getter(AccessLevel.PROTECTED) + @Nullable private final Supplier<SeekableDataInputStream> inputStreamSupplier; // Toggle flag, whether to read blocks lazily (I/O intensive) or not (Memory intensive) + @Getter(AccessLevel.NONE) protected boolean readBlockLazily; - public HoodieLogBlock( - @Nonnull Map<HeaderMetadataType, String> logBlockHeader, - @Nonnull Map<FooterMetadataType, String> logBlockFooter, - @Nonnull Option<HoodieLogBlockContentLocation> blockContentLocation, - @Nonnull Option<byte[]> content, - @Nullable Supplier<SeekableDataInputStream> inputStreamSupplier, - boolean readBlockLazily) { - this.logBlockHeader = logBlockHeader; - this.logBlockFooter = logBlockFooter; - this.blockContentLocation = blockContentLocation; - this.content = content; - this.inputStreamSupplier = inputStreamSupplier; - this.readBlockLazily = readBlockLazily; - } - // Return the bytes representation of the data belonging to a LogBlock public ByteArrayOutputStream getContentBytes(HoodieStorage storage) throws IOException { throw new HoodieException("No implementation was provided"); @@ -110,22 +107,6 @@ public abstract class HoodieLogBlock { throw new HoodieException("No implementation was provided"); } - public Option<HoodieLogBlockContentLocation> getBlockContentLocation() { - return this.blockContentLocation; - } - - public Map<HeaderMetadataType, String> getLogBlockHeader() { - return logBlockHeader; - } - - public Map<FooterMetadataType, String> getLogBlockFooter() { - return logBlockFooter; - } - - public Option<byte[]> getContent() { - return content; - } - /** * Compacted blocks are created using log compaction which basically merges the consecutive blocks together and create * huge block with all the changes. @@ -161,10 +142,10 @@ public abstract class HoodieLogBlock { try { logBlockHeader.put(HeaderMetadataType.RECORD_POSITIONS, LogReaderUtils.encodePositions(positionSet)); } catch (IOException e) { - LOG.error("Cannot write record positions to the log block header.", e); + log.error("Cannot write record positions to the log block header.", e); } } else { - LOG.warn("There are duplicate keys in the records (number of unique positions: {}, " + log.warn("There are duplicate keys in the records (number of unique positions: {}, " + "number of records: {}). Skip writing record positions to the log block header.", positionSet.size(), numRecords); } @@ -176,7 +157,7 @@ public abstract class HoodieLogBlock { } protected void removeBaseFileInstantTimeOfPositions() { - LOG.info("There are records without valid positions. " + log.info("There are records without valid positions. " + "Skip writing record positions to the block header."); logBlockHeader.remove(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS); } @@ -252,7 +233,10 @@ public abstract class HoodieLogBlock { * This class is used to store the Location of the Content of a Log Block. It's used when a client chooses for a IO * intensive CompactedScanner, the location helps to lazily read contents from the log file */ + @AllArgsConstructor + @Getter public static final class HoodieLogBlockContentLocation { + // Storage Config required to access the file private final HoodieStorage storage; // The logFile that contains this block @@ -263,38 +247,6 @@ public abstract class HoodieLogBlock { private final long blockSize; // The final position where the complete block ends private final long blockEndPos; - - public HoodieLogBlockContentLocation(HoodieStorage storage, - HoodieLogFile logFile, - long contentPositionInLogFile, - long blockSize, - long blockEndPos) { - this.storage = storage; - this.logFile = logFile; - this.contentPositionInLogFile = contentPositionInLogFile; - this.blockSize = blockSize; - this.blockEndPos = blockEndPos; - } - - public HoodieStorage getStorage() { - return storage; - } - - public HoodieLogFile getLogFile() { - return logFile; - } - - public long getContentPositionInLogFile() { - return contentPositionInLogFile; - } - - public long getBlockSize() { - return blockSize; - } - - public long getBlockEndPos() { - return blockEndPos; - } } /** @@ -359,10 +311,6 @@ public abstract class HoodieLogBlock { return Option.of(baos); } - protected Supplier<SeekableDataInputStream> getInputStreamSupplier() { - return inputStreamSupplier; - } - /** * Adds the record positions if the base file instant time of the positions exists * in the log header and the record positions are all valid. diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecord.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecord.java index d68e608463df..fb78adaa772c 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecord.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecord.java @@ -23,6 +23,10 @@ import org.apache.hudi.common.model.HoodieOperation; import org.apache.hudi.common.schema.HoodieSchema; import org.apache.hudi.common.util.OrderingValues; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.Setter; + import javax.annotation.Nullable; import java.io.Serializable; @@ -34,41 +38,22 @@ import java.util.function.UnaryOperator; * * @param <T> The type of the engine specific row. */ +@AllArgsConstructor +@Getter public class BufferedRecord<T> implements Serializable { + private String recordKey; - private T record; private final Comparable orderingValue; + private T record; private final Integer schemaId; - @Nullable private HoodieOperation hoodieOperation; + @Nullable + @Setter + private HoodieOperation hoodieOperation; public BufferedRecord() { this(null, null, null, null, null); } - public BufferedRecord(String recordKey, Comparable orderingValue, T record, Integer schemaId, @Nullable HoodieOperation hoodieOperation) { - this.recordKey = recordKey; - this.orderingValue = orderingValue; - this.record = record; - this.schemaId = schemaId; - this.hoodieOperation = hoodieOperation; - } - - public String getRecordKey() { - return recordKey; - } - - public Comparable getOrderingValue() { - return orderingValue; - } - - public T getRecord() { - return record; - } - - public Integer getSchemaId() { - return schemaId; - } - public boolean isDelete() { return HoodieOperation.isDelete(hoodieOperation) || HoodieOperation.isUpdateBefore(hoodieOperation); } @@ -81,14 +66,6 @@ public class BufferedRecord<T> implements Serializable { return isDelete() && OrderingValues.isDefault(orderingValue); } - public void setHoodieOperation(HoodieOperation hoodieOperation) { - this.hoodieOperation = hoodieOperation; - } - - public HoodieOperation getHoodieOperation() { - return this.hoodieOperation; - } - public BufferedRecord<T> toBinary(RecordContext<T> recordContext) { if (record != null) { HoodieSchema schema = recordContext.getSchemaFromBufferRecord(this); @@ -124,6 +101,8 @@ public class BufferedRecord<T> implements Serializable { return this; } + // Intentionally not using @EqualsAndHashCode: Lombok generates instanceof/canEqual based equality, + // while this class requires exact runtime-class equality via getClass() @Override public boolean equals(Object o) { if (this == o) { diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java index 81d2de0ac0a2..06e903dd1fa4 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java @@ -33,16 +33,17 @@ import org.apache.hudi.common.util.OrderingValues; import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.exception.HoodieIOException; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + import java.io.IOException; /** * Factory to create a {@link BufferedRecordMerger}. */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) public class BufferedRecordMergerFactory { - private BufferedRecordMergerFactory() { - } - public static <T> BufferedRecordMerger<T> create(HoodieReaderContext<T> readerContext, RecordMergeMode recordMergeMode, boolean enablePartialMerging, diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/DeleteContext.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/DeleteContext.java index ce6857dc2aeb..f1bd8b0b0a8a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/DeleteContext.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/DeleteContext.java @@ -27,6 +27,9 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.StringUtils; import org.apache.hudi.common.util.collection.Pair; +import lombok.Getter; +import lombok.experimental.Accessors; + import java.io.Serializable; import java.util.Properties; @@ -37,10 +40,13 @@ import static org.apache.hudi.common.model.HoodieRecord.HOODIE_IS_DELETED_FIELD; /** * Schema context for deletes. */ +@Getter public class DeleteContext implements Serializable { + private static final long serialVersionUID = 1L; private final Option<Pair<String, String>> customDeleteMarkerKeyValue; + @Accessors(fluent = true) private final boolean hasBuiltInDeleteField; private int hoodieOperationPos; private HoodieSchema readerSchema; @@ -108,25 +114,9 @@ public class DeleteContext implements Serializable { .orElseGet(() -> -1); } - public Option<Pair<String, String>> getCustomDeleteMarkerKeyValue() { - return customDeleteMarkerKeyValue; - } - - public boolean hasBuiltInDeleteField() { - return hasBuiltInDeleteField; - } - - public int getHoodieOperationPos() { - return hoodieOperationPos; - } - public DeleteContext withReaderSchema(HoodieSchema readerSchema) { this.readerSchema = readerSchema; this.hoodieOperationPos = getHoodieOperationPos(readerSchema); return this; } - - public HoodieSchema getReaderSchema() { - return readerSchema; - } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java index 01f3e513e341..da0bfb567abe 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java @@ -32,8 +32,11 @@ import org.apache.hudi.common.util.VisibleForTesting; import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.exception.HoodieException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; import javax.annotation.Nullable; @@ -108,11 +111,13 @@ import java.util.stream.Stream; * * <p>IMPORTANT: the reader may optionally choose to fall back to reading the latest snapshot if there are files decoding the commit metadata are already cleaned. */ +@Slf4j public class IncrementalQueryAnalyzer { + public static final String START_COMMIT_EARLIEST = "earliest"; - private static final Logger LOG = LoggerFactory.getLogger(IncrementalQueryAnalyzer.class); private final HoodieTableMetaClient metaClient; + @Getter private final Option<String> startCompletionTime; private final Option<String> endCompletionTime; private final InstantRange.RangeType rangeType; @@ -143,10 +148,6 @@ public class IncrementalQueryAnalyzer { this.limit = limit; } - public Option<String> getStartCompletionTime() { - return startCompletionTime; - } - /** * Returns a builder. */ @@ -206,7 +207,7 @@ public class IncrementalQueryAnalyzer { String endInstant = endCompletionTime.isEmpty() ? null : lastInstant; return QueryContext.create(startInstant, endInstant, instants, archivedInstants, activeInstants, filteredTimeline, archivedReadTimeline); } catch (Exception ex) { - LOG.error("Got exception when generating incremental query info", ex); + log.error("Got exception when generating incremental query info", ex); throw new HoodieException(ex); } } @@ -284,6 +285,7 @@ public class IncrementalQueryAnalyzer { /** * Builder for {@link IncrementalQueryAnalyzer}. */ + @NoArgsConstructor public static class Builder { /** * Start completion time. @@ -304,9 +306,6 @@ public class IncrementalQueryAnalyzer { */ private int limit = -1; - public Builder() { - } - public Builder startCompletionTime(String startCompletionTime) { this.startCompletionTime = startCompletionTime; return this; @@ -361,9 +360,12 @@ public class IncrementalQueryAnalyzer { /** * Represents the analyzed query context. */ + @AllArgsConstructor(access = AccessLevel.PRIVATE) + @Getter public static class QueryContext { + public static final QueryContext EMPTY = - new QueryContext(null, null, Collections.emptyList(), Collections.emptyList(), Collections.emptyList(), null, null); + new QueryContext(Option.empty(), Option.empty(), Collections.emptyList(), Collections.emptyList(), Collections.emptyList(), null, null); /** * An empty option indicates consumption from the earliest instant. @@ -373,6 +375,8 @@ public class IncrementalQueryAnalyzer { * An empty option indicates consumption to the latest instant. */ private final Option<String> endInstant; + @Getter(AccessLevel.NONE) + private final List<String> instants; private final List<HoodieInstant> archivedInstants; private final List<HoodieInstant> activeInstants; /** @@ -382,25 +386,9 @@ public class IncrementalQueryAnalyzer { /** * The archived timeline to read filtered by given configurations. */ + @Nullable + @Getter(AccessLevel.NONE) private final HoodieTimeline archivedTimeline; - private final List<String> instants; - - private QueryContext( - @Nullable String startInstant, - @Nullable String endInstant, - List<String> instants, - List<HoodieInstant> archivedInstants, - List<HoodieInstant> activeInstants, - HoodieTimeline activeTimeline, - @Nullable HoodieTimeline archivedTimeline) { - this.startInstant = Option.ofNullable(startInstant); - this.endInstant = Option.ofNullable(endInstant); - this.archivedInstants = archivedInstants; - this.activeInstants = activeInstants; - this.activeTimeline = activeTimeline; - this.archivedTimeline = archivedTimeline; - this.instants = instants; - } public static QueryContext create( @Nullable String startInstant, @@ -410,7 +398,7 @@ public class IncrementalQueryAnalyzer { List<HoodieInstant> activeInstants, HoodieTimeline activeTimeline, @Nullable HoodieTimeline archivedTimeline) { - return new QueryContext(startInstant, endInstant, instants, archivedInstants, activeInstants, activeTimeline, archivedTimeline); + return new QueryContext(Option.ofNullable(startInstant), Option.ofNullable(endInstant), instants, archivedInstants, activeInstants, activeTimeline, archivedTimeline); } public boolean isEmpty() { @@ -421,14 +409,6 @@ public class IncrementalQueryAnalyzer { return this.instants; } - public Option<String> getStartInstant() { - return startInstant; - } - - public Option<String> getEndInstant() { - return endInstant; - } - /** * Returns the latest instant time which should be included physically in reading. */ @@ -441,14 +421,6 @@ public class IncrementalQueryAnalyzer { return Stream.concat(archivedInstants.stream(), activeInstants.stream()).collect(Collectors.toList()); } - public List<HoodieInstant> getArchivedInstants() { - return archivedInstants; - } - - public List<HoodieInstant> getActiveInstants() { - return activeInstants; - } - public boolean isConsumingFromEarliest() { return startInstant.isEmpty(); } @@ -489,10 +461,6 @@ public class IncrementalQueryAnalyzer { } } - public HoodieTimeline getActiveTimeline() { - return this.activeTimeline; - } - public @Nullable HoodieTimeline getArchivedTimeline() { return archivedTimeline; } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java index e84921c080ae..e8ff9add0ad8 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/DefaultFileGroupRecordBufferLoader.java @@ -33,6 +33,9 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.storage.HoodieStorage; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + import java.util.List; /** @@ -40,16 +43,15 @@ import java.util.List; * * @param <T> the engine specific record type */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) class DefaultFileGroupRecordBufferLoader<T> extends LogScanningRecordBufferLoader implements FileGroupRecordBufferLoader<T> { + private static final DefaultFileGroupRecordBufferLoader INSTANCE = new DefaultFileGroupRecordBufferLoader<>(); static <T> DefaultFileGroupRecordBufferLoader<T> getInstance() { return INSTANCE; } - private DefaultFileGroupRecordBufferLoader() { - } - @Override public Pair<HoodieFileGroupRecordBuffer<T>, List<String>> getRecordBuffer(HoodieReaderContext<T> readerContext, HoodieStorage storage, diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java index 702ecf5041e0..8d5d967ec256 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java @@ -50,6 +50,9 @@ import org.apache.hudi.internal.schema.InternalSchema; import org.apache.hudi.internal.schema.action.InternalSchemaMerger; import org.apache.hudi.internal.schema.convert.InternalSchemaConverter; +import lombok.Getter; +import lombok.Setter; + import java.io.IOException; import java.io.Serializable; import java.util.Iterator; @@ -77,8 +80,10 @@ abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordBuffer<T protected final Option<Pair<String, String>> payloadClasses; protected final TypedProperties props; protected final ExternalSpillableMap<Serializable, BufferedRecord<T>> records; + @Getter protected final DeleteContext deleteContext; protected final BufferedRecordConverter<T> bufferedRecordConverter; + @Setter protected ClosableIterator<T> baseFileIterator; protected UpdateProcessor<T> updateProcessor; protected Iterator<BufferedRecord<T>> logRecordIterator; @@ -87,6 +92,7 @@ abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordBuffer<T protected InternalSchema internalSchema; protected HoodieTableMetaClient hoodieTableMetaClient; protected BufferedRecordMerger<T> bufferedRecordMerger; + @Getter protected long totalLogRecords = 0; protected FileGroupRecordBuffer(HoodieReaderContext<T> readerContext, @@ -131,15 +137,6 @@ abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordBuffer<T readerContext.getRecordSizeEstimator(), diskMapType, readerContext.getRecordSerializer(), isBitCaskDiskMapCompressionEnabled, getClass().getSimpleName()); } - @Override - public void setBaseFileIterator(ClosableIterator<T> baseFileIterator) { - this.baseFileIterator = baseFileIterator; - } - - public DeleteContext getDeleteContext() { - return deleteContext; - } - /** * This allows hasNext() to be called multiple times without incrementing the iterator by more than 1 * record. It does come with the caveat that hasNext() must be called every time before next(). But @@ -169,10 +166,6 @@ abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordBuffer<T return records.size(); } - public long getTotalLogRecords() { - return totalLogRecords; - } - @Override public ClosableIterator<BufferedRecord<T>> getLogRecordIterator() { return new LogRecordIterator<>(this); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java index 73dd5dda5e87..5eaa54e033c8 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java @@ -41,9 +41,8 @@ import org.apache.hudi.common.util.collection.ClosableIterator; import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.exception.HoodieKeyException; +import lombok.extern.slf4j.Slf4j; import org.roaringbitmap.longlong.Roaring64NavigableMap; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.Serializable; @@ -61,8 +60,8 @@ import java.util.function.Function; * Here the position means that record position in the base file. The records from the base file is accessed from an iterator object. These records are merged when the * {@link #hasNext} method is called. */ +@Slf4j public class PositionBasedFileGroupRecordBuffer<T> extends KeyBasedFileGroupRecordBuffer<T> { - private static final Logger LOG = LoggerFactory.getLogger(PositionBasedFileGroupRecordBuffer.class); private static final String ROW_INDEX_COLUMN_NAME = "row_index"; public static final String ROW_INDEX_TEMPORARY_COLUMN_NAME = "_tmp_metadata_" + ROW_INDEX_COLUMN_NAME; @@ -96,7 +95,7 @@ public class PositionBasedFileGroupRecordBuffer<T> extends KeyBasedFileGroupReco // Extract positions from data block. List<Long> recordPositions = extractRecordPositions(dataBlock, baseFileInstantTime); if (recordPositions == null) { - LOG.debug("Falling back to key based merge for data block"); + log.debug("Falling back to key based merge for data block"); fallbackToKeyBasedBuffer(); super.processDataBlock(dataBlock, keySpecOpt); return; @@ -181,7 +180,7 @@ public class PositionBasedFileGroupRecordBuffer<T> extends KeyBasedFileGroupReco List<Long> recordPositions = extractRecordPositions(deleteBlock, baseFileInstantTime); if (recordPositions == null) { - LOG.debug("Falling back to key based merging for delete block"); + log.debug("Falling back to key based merging for delete block"); fallbackToKeyBasedBuffer(); super.processDeleteBlock(deleteBlock); return; @@ -291,7 +290,7 @@ public class PositionBasedFileGroupRecordBuffer<T> extends KeyBasedFileGroupReco String blockBaseFileInstantTime = logBlock.getBaseFileInstantTimeOfPositions(); if (StringUtils.isNullOrEmpty(blockBaseFileInstantTime) || !baseFileInstantTime.equals(blockBaseFileInstantTime)) { - LOG.debug("The record positions cannot be used because the base file instant time " + log.debug("The record positions cannot be used because the base file instant time " + "is either missing or different from the base file to merge. " + "Instant time in the header: {}, base file instant time of the file group: {}.", blockBaseFileInstantTime, baseFileInstantTime); @@ -299,7 +298,7 @@ public class PositionBasedFileGroupRecordBuffer<T> extends KeyBasedFileGroupReco } Roaring64NavigableMap positions = logBlock.getRecordPositions(); if (positions == null || positions.isEmpty()) { - LOG.info("No record position info is found when attempting to do position based merge."); + log.info("No record position info is found when attempting to do position based merge."); return null; } @@ -309,7 +308,7 @@ public class PositionBasedFileGroupRecordBuffer<T> extends KeyBasedFileGroupReco } if (blockPositions.isEmpty()) { - LOG.info("No positions are extracted."); + log.info("No positions are extracted."); return null; } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitSerializer.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitSerializer.java index 7f6771ba3940..57b0a30c4cb1 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitSerializer.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitSerializer.java @@ -103,13 +103,13 @@ public class HoodieSourceSplitSerializer implements SimpleVersionedSerializer<Ho out.writeBoolean(false); } - out.writeBoolean(instantRange.getStartInstant().isPresent()); - if (instantRange.getStartInstant().isPresent()) { - out.writeUTF(instantRange.getStartInstant().get()); + out.writeBoolean(instantRange.getStartInstantOpt().isPresent()); + if (instantRange.getStartInstantOpt().isPresent()) { + out.writeUTF(instantRange.getStartInstantOpt().get()); } - out.writeBoolean(instantRange.getEndInstant().isPresent()); - if (instantRange.getEndInstant().isPresent()) { - out.writeUTF(instantRange.getEndInstant().get()); + out.writeBoolean(instantRange.getEndInstantOpt().isPresent()); + if (instantRange.getEndInstantOpt().isPresent()) { + out.writeUTF(instantRange.getEndInstantOpt().get()); } if (instantRange.getRangeType().equals(InstantRange.RangeType.EXACT_MATCH)) { diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java index deaa492ce15d..ce97b2e70f75 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java @@ -438,11 +438,11 @@ public class TestIncrementalInputSplits extends HoodieCommonTestHarness { IncrementalInputSplits.Result result = iis.inputSplits(metaClient, firstInstant.getCompletionTime(), false); String minStartCommit = result.getInputSplits().stream() - .map(split -> split.getInstantRange().get().getStartInstant().get()) + .map(split -> split.getInstantRange().get().getStartInstantOpt().get()) .min((commit1,commit2) -> compareTimestamps(commit1, LESSER_THAN, commit2) ? 1 : 0) .orElse(null); String maxEndCommit = result.getInputSplits().stream() - .map(split -> split.getInstantRange().get().getEndInstant().get()) + .map(split -> split.getInstantRange().get().getEndInstantOpt().get()) .max((commit1,commit2) -> compareTimestamps(commit1, GREATER_THAN, commit2) ? 1 : 0) .orElse(null); assertEquals(0, intervalBetween2Instants(commitsTimeline, minStartCommit, maxEndCommit), "Should read 1 instant"); diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java index ec6f3b863caf..bb28ee92e2df 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java @@ -505,8 +505,8 @@ public class TestStreamReadMonitoringFunction { private static boolean isPointInstantRange(InstantRange instantRange, String timestamp) { return instantRange != null - && Objects.equals(timestamp, instantRange.getStartInstant().get()) - && Objects.equals(timestamp, instantRange.getEndInstant().get()); + && Objects.equals(timestamp, instantRange.getStartInstantOpt().get()) + && Objects.equals(timestamp, instantRange.getEndInstantOpt().get()); } private AbstractStreamOperatorTestHarness<MergeOnReadInputSplit> createHarness( diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplit.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplit.java index 1cb244245ce8..8dd32af3adf3 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplit.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplit.java @@ -360,8 +360,8 @@ public class TestHoodieSourceSplit { ); assertTrue(split.getInstantRange().isPresent()); - assertEquals("20230101000000000", split.getInstantRange().get().getStartInstant().get()); - assertEquals("20230131235959999", split.getInstantRange().get().getEndInstant().get()); + assertEquals("20230101000000000", split.getInstantRange().get().getStartInstantOpt().get()); + assertEquals("20230131235959999", split.getInstantRange().get().getEndInstantOpt().get()); } @Test @@ -402,9 +402,9 @@ public class TestHoodieSourceSplit { ); assertTrue(split.getInstantRange().isPresent()); - assertTrue(split.getInstantRange().get().getStartInstant().isPresent()); - assertFalse(split.getInstantRange().get().getEndInstant().isPresent()); - assertEquals("20230101000000000", split.getInstantRange().get().getStartInstant().get()); + assertTrue(split.getInstantRange().get().getStartInstantOpt().isPresent()); + assertFalse(split.getInstantRange().get().getEndInstantOpt().isPresent()); + assertEquals("20230101000000000", split.getInstantRange().get().getStartInstantOpt().get()); } @Test @@ -476,7 +476,7 @@ public class TestHoodieSourceSplit { ); assertTrue(split.getInstantRange().isPresent()); - assertTrue(split.getInstantRange().get().getStartInstant().isPresent()); - assertTrue(split.getInstantRange().get().getEndInstant().isPresent()); + assertTrue(split.getInstantRange().get().getStartInstantOpt().isPresent()); + assertTrue(split.getInstantRange().get().getEndInstantOpt().isPresent()); } } diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplitSerializer.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplitSerializer.java index fc30da2193d7..04b526a80542 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplitSerializer.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestHoodieSourceSplitSerializer.java @@ -509,10 +509,10 @@ public class TestHoodieSourceSplitSerializer { assertNotNull(deserialized); assertTrue(deserialized.getInstantRange().isPresent()); - assertTrue(deserialized.getInstantRange().get().getStartInstant().isPresent()); - assertTrue(deserialized.getInstantRange().get().getEndInstant().isPresent()); - assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstant().get()); - assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstant().get()); + assertTrue(deserialized.getInstantRange().get().getStartInstantOpt().isPresent()); + assertTrue(deserialized.getInstantRange().get().getEndInstantOpt().isPresent()); + assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstantOpt().get()); + assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstantOpt().get()); } @Test @@ -540,9 +540,9 @@ public class TestHoodieSourceSplitSerializer { assertNotNull(deserialized); assertTrue(deserialized.getInstantRange().isPresent()); - assertTrue(deserialized.getInstantRange().get().getStartInstant().isPresent()); - assertFalse(deserialized.getInstantRange().get().getEndInstant().isPresent()); - assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstant().get()); + assertTrue(deserialized.getInstantRange().get().getStartInstantOpt().isPresent()); + assertFalse(deserialized.getInstantRange().get().getEndInstantOpt().isPresent()); + assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstantOpt().get()); } @Test @@ -570,10 +570,10 @@ public class TestHoodieSourceSplitSerializer { assertNotNull(deserialized); assertTrue(deserialized.getInstantRange().isPresent()); - assertTrue(deserialized.getInstantRange().get().getStartInstant().isPresent()); - assertTrue(deserialized.getInstantRange().get().getEndInstant().isPresent()); - assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstant().get()); - assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstant().get()); + assertTrue(deserialized.getInstantRange().get().getStartInstantOpt().isPresent()); + assertTrue(deserialized.getInstantRange().get().getEndInstantOpt().isPresent()); + assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstantOpt().get()); + assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstantOpt().get()); } @Test @@ -605,8 +605,8 @@ public class TestHoodieSourceSplitSerializer { assertTrue(deserialized.getInstantRange().isPresent()); assertEquals(10, deserialized.getFileOffset()); assertEquals(500L, deserialized.getConsumed()); - assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstant().get()); - assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstant().get()); + assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstantOpt().get()); + assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstantOpt().get()); } @Test @@ -637,13 +637,13 @@ public class TestHoodieSourceSplitSerializer { // Verify split1 assertTrue(deserialized1.getInstantRange().isPresent()); - assertEquals("20230101000000000", deserialized1.getInstantRange().get().getStartInstant().get()); - assertEquals("20230131235959999", deserialized1.getInstantRange().get().getEndInstant().get()); + assertEquals("20230101000000000", deserialized1.getInstantRange().get().getStartInstantOpt().get()); + assertEquals("20230131235959999", deserialized1.getInstantRange().get().getEndInstantOpt().get()); // Verify split2 assertTrue(deserialized2.getInstantRange().isPresent()); - assertEquals("20230201000000000", deserialized2.getInstantRange().get().getStartInstant().get()); - assertFalse(deserialized2.getInstantRange().get().getEndInstant().isPresent()); + assertEquals("20230201000000000", deserialized2.getInstantRange().get().getStartInstantOpt().get()); + assertFalse(deserialized2.getInstantRange().get().getEndInstantOpt().isPresent()); // Verify split3 assertFalse(deserialized3.getInstantRange().isPresent()); @@ -863,15 +863,15 @@ public class TestHoodieSourceSplitSerializer { assertTrue(deserialized2.getInstantRange().isPresent()); assertEquals(InstantRange.RangeType.OPEN_CLOSED, deserialized2.getInstantRange().get().getRangeType()); - assertEquals("20230201000000000", deserialized2.getInstantRange().get().getStartInstant().get()); - assertEquals("20230228235959999", deserialized2.getInstantRange().get().getEndInstant().get()); + assertEquals("20230201000000000", deserialized2.getInstantRange().get().getStartInstantOpt().get()); + assertEquals("20230228235959999", deserialized2.getInstantRange().get().getEndInstantOpt().get()); // Verify split3 (CLOSED_CLOSED) assertTrue(deserialized3.getInstantRange().isPresent()); assertEquals(InstantRange.RangeType.CLOSED_CLOSED, deserialized3.getInstantRange().get().getRangeType()); - assertEquals("20230301000000000", deserialized3.getInstantRange().get().getStartInstant().get()); - assertEquals("20230331235959999", deserialized3.getInstantRange().get().getEndInstant().get()); + assertEquals("20230301000000000", deserialized3.getInstantRange().get().getStartInstantOpt().get()); + assertEquals("20230331235959999", deserialized3.getInstantRange().get().getEndInstantOpt().get()); } @Test @@ -954,9 +954,9 @@ public class TestHoodieSourceSplitSerializer { assertTrue(deserialized.getInstantRange().isPresent()); assertEquals(InstantRange.RangeType.CLOSED_CLOSED, deserialized.getInstantRange().get().getRangeType()); - assertTrue(deserialized.getInstantRange().get().getStartInstant().isPresent()); - assertFalse(deserialized.getInstantRange().get().getEndInstant().isPresent()); - assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstant().get()); + assertTrue(deserialized.getInstantRange().get().getStartInstantOpt().isPresent()); + assertFalse(deserialized.getInstantRange().get().getEndInstantOpt().isPresent()); + assertEquals("20230101000000000", deserialized.getInstantRange().get().getStartInstantOpt().get()); // Verify range behavior - start is inclusive, no end boundary assertTrue(deserialized.getInstantRange().get().isInRange("20230101000000000")); // start inclusive @@ -992,9 +992,9 @@ public class TestHoodieSourceSplitSerializer { assertTrue(deserialized.getInstantRange().isPresent()); assertEquals(InstantRange.RangeType.CLOSED_CLOSED, deserialized.getInstantRange().get().getRangeType()); - assertFalse(deserialized.getInstantRange().get().getStartInstant().isPresent()); - assertTrue(deserialized.getInstantRange().get().getEndInstant().isPresent()); - assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstant().get()); + assertFalse(deserialized.getInstantRange().get().getStartInstantOpt().isPresent()); + assertTrue(deserialized.getInstantRange().get().getEndInstantOpt().isPresent()); + assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstantOpt().get()); // Verify range behavior - no start boundary, end is inclusive assertTrue(deserialized.getInstantRange().get().isInRange("19700101000000000")); @@ -1029,9 +1029,9 @@ public class TestHoodieSourceSplitSerializer { assertTrue(deserialized.getInstantRange().isPresent()); assertEquals(InstantRange.RangeType.OPEN_CLOSED, deserialized.getInstantRange().get().getRangeType()); - assertFalse(deserialized.getInstantRange().get().getStartInstant().isPresent()); - assertTrue(deserialized.getInstantRange().get().getEndInstant().isPresent()); - assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstant().get()); + assertFalse(deserialized.getInstantRange().get().getStartInstantOpt().isPresent()); + assertTrue(deserialized.getInstantRange().get().getEndInstantOpt().isPresent()); + assertEquals("20230131235959999", deserialized.getInstantRange().get().getEndInstantOpt().get()); // Verify range behavior - no start boundary, end is inclusive assertTrue(deserialized.getInstantRange().get().isInRange("19700101000000000")); @@ -1087,23 +1087,23 @@ public class TestHoodieSourceSplitSerializer { HoodieSourceSplit deserialized4 = serializer.deserialize(serializer.getVersion(), serialized4); // Verify OPEN_CLOSED with only start - assertTrue(deserialized1.getInstantRange().get().getStartInstant().isPresent()); - assertFalse(deserialized1.getInstantRange().get().getEndInstant().isPresent()); + assertTrue(deserialized1.getInstantRange().get().getStartInstantOpt().isPresent()); + assertFalse(deserialized1.getInstantRange().get().getEndInstantOpt().isPresent()); assertEquals(InstantRange.RangeType.OPEN_CLOSED, deserialized1.getInstantRange().get().getRangeType()); // Verify OPEN_CLOSED with only end - assertFalse(deserialized2.getInstantRange().get().getStartInstant().isPresent()); - assertTrue(deserialized2.getInstantRange().get().getEndInstant().isPresent()); + assertFalse(deserialized2.getInstantRange().get().getStartInstantOpt().isPresent()); + assertTrue(deserialized2.getInstantRange().get().getEndInstantOpt().isPresent()); assertEquals(InstantRange.RangeType.OPEN_CLOSED, deserialized2.getInstantRange().get().getRangeType()); // Verify CLOSED_CLOSED with only start - assertTrue(deserialized3.getInstantRange().get().getStartInstant().isPresent()); - assertFalse(deserialized3.getInstantRange().get().getEndInstant().isPresent()); + assertTrue(deserialized3.getInstantRange().get().getStartInstantOpt().isPresent()); + assertFalse(deserialized3.getInstantRange().get().getEndInstantOpt().isPresent()); assertEquals(InstantRange.RangeType.CLOSED_CLOSED, deserialized3.getInstantRange().get().getRangeType()); // Verify CLOSED_CLOSED with only end - assertFalse(deserialized4.getInstantRange().get().getStartInstant().isPresent()); - assertTrue(deserialized4.getInstantRange().get().getEndInstant().isPresent()); + assertFalse(deserialized4.getInstantRange().get().getStartInstantOpt().isPresent()); + assertTrue(deserialized4.getInstantRange().get().getEndInstantOpt().isPresent()); assertEquals(InstantRange.RangeType.CLOSED_CLOSED, deserialized4.getInstantRange().get().getRangeType()); }
