Skip to content

[HUDI-1552] Improve performance of key lookups from base file in Metadata Table. #2494

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vinothchandar merged 2 commits into from
Mar 15, 2021
Merged

[HUDI-1552] Improve performance of key lookups from base file in Metadata Table. #2494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
874cc6d
[HUDI-1552] Improve performance of key lookups from base file in Meta…
prashantwason Jan 26, 2021
59b919a
Properly close the readers, when metadata table is accessed from exec…
vinothchandar Mar 14, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions hudi-cli/src/main/java/org/apache/hudi/cli/commands/MetadataCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.hudi.cli.utils.SparkUtil;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.config.HoodieWriteConfig;
Expand Down Expand Up @@ -147,7 +148,8 @@ public String init(@CliOption(key = {"readonly"}, unspecifiedDefaultValue = "fal
public String stats() throws IOException {
HoodieCLI.getTableMetaClient();
HoodieMetadataConfig config = HoodieMetadataConfig.newBuilder().enable(true).build();
HoodieBackedTableMetadata metadata = new HoodieBackedTableMetadata(HoodieCLI.conf, config, HoodieCLI.basePath, "/tmp");
HoodieBackedTableMetadata metadata = new HoodieBackedTableMetadata(new HoodieLocalEngineContext(HoodieCLI.conf),
config, HoodieCLI.basePath, "/tmp");
Map<String, String> stats = metadata.stats();

StringBuffer out = new StringBuffer("\n");
Expand Down Expand Up @@ -197,7 +199,7 @@ public String listFiles(
final String partition) throws IOException {
HoodieCLI.getTableMetaClient();
HoodieMetadataConfig config = HoodieMetadataConfig.newBuilder().enable(true).build();
HoodieBackedTableMetadata metaReader = new HoodieBackedTableMetadata(HoodieCLI.conf, config, HoodieCLI.basePath, "/tmp");
HoodieBackedTableMetadata metaReader = new HoodieBackedTableMetadata(new HoodieLocalEngineContext(HoodieCLI.conf), config, HoodieCLI.basePath, "/tmp");

StringBuffer out = new StringBuffer("\n");
if (!metaReader.enabled()) {
Expand Down
3 changes: 1 addition & 2 deletions hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,9 +107,8 @@ protected HoodieTable(HoodieWriteConfig config, HoodieEngineContext context, Hoo
this.hadoopConfiguration = context.getHadoopConf();
this.context = context;

// disable reuse of resources, given there is no close() called on the executors ultimately
HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder().fromProperties(config.getMetadataConfig().getProps())
.enableReuse(false).build();
.build();
this.metadata = HoodieTableMetadata.create(context, metadataConfig, config.getBasePath(),
FileSystemViewStorageConfig.DEFAULT_VIEW_SPILLABLE_DIR);

Expand Down
1 change: 0 additions & 1 deletion ...nt/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -973,7 +973,6 @@ private HoodieWriteConfig.Builder getWriteConfigBuilder(boolean autoCommit, bool
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BLOOM).build())
.withMetadataConfig(HoodieMetadataConfig.newBuilder()
.enable(useFileListingMetadata)
.enableReuse(false)
.enableMetrics(enableMetrics)
.enableFallback(false).build())
.withMetricsConfig(HoodieMetricsConfig.newBuilder().on(enableMetrics)
Expand Down
15 changes: 0 additions & 15 deletions hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,10 +67,6 @@ public final class HoodieMetadataConfig extends DefaultHoodieConfig {
public static final String CLEANER_COMMITS_RETAINED_PROP = METADATA_PREFIX + ".cleaner.commits.retained";
public static final int DEFAULT_CLEANER_COMMITS_RETAINED = 3;

// Controls whether or no the base file open/log merges are reused per API call
public static final String ENABLE_REUSE_PROP = METADATA_PREFIX + ".reuse.enable";
public static final String DEFAULT_ENABLE_REUSE = "true";

// Controls whether or not, upon failure to fetch from metadata table, should fallback to listing.
public static final String ENABLE_FALLBACK_PROP = METADATA_PREFIX + ".fallback.enable";
public static final String DEFAULT_ENABLE_FALLBACK = "true";
Expand Down Expand Up @@ -105,10 +101,6 @@ public boolean useFileListingMetadata() {
return Boolean.parseBoolean(props.getProperty(METADATA_ENABLE_PROP));
}

public boolean enableReuse() {
return Boolean.parseBoolean(props.getProperty(ENABLE_REUSE_PROP));
}

public boolean enableFallback() {
return Boolean.parseBoolean(props.getProperty(ENABLE_FALLBACK_PROP));
}
Expand Down Expand Up @@ -151,11 +143,6 @@ public Builder enableMetrics(boolean enableMetrics) {
return this;
}

public Builder enableReuse(boolean reuse) {
props.setProperty(ENABLE_REUSE_PROP, String.valueOf(reuse));
return this;
}

public Builder enableFallback(boolean fallback) {
props.setProperty(ENABLE_FALLBACK_PROP, String.valueOf(fallback));
return this;
Expand Down Expand Up @@ -233,8 +220,6 @@ public HoodieMetadataConfig build() {
HOODIE_ASSUME_DATE_PARTITIONING_PROP, DEFAULT_ASSUME_DATE_PARTITIONING);
setDefaultOnCondition(props, !props.containsKey(ENABLE_FALLBACK_PROP), ENABLE_FALLBACK_PROP,
DEFAULT_ENABLE_FALLBACK);
setDefaultOnCondition(props, !props.containsKey(ENABLE_REUSE_PROP), ENABLE_REUSE_PROP,
DEFAULT_ENABLE_REUSE);
setDefaultOnCondition(props, !props.containsKey(DIRECTORY_FILTER_REGEX), DIRECTORY_FILTER_REGEX,
DEFAULT_DIRECTORY_FILTER_REGEX);
return config;
Expand Down
2 changes: 1 addition & 1 deletion hudi-common/src/main/java/org/apache/hudi/common/table/view/FileSystemViewManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,7 +214,7 @@ public static FileSystemViewManager createViewManager(final HoodieEngineContext
final FileSystemViewStorageConfig config,
final String basePath) {
return createViewManager(context, metadataConfig, config,
() -> HoodieTableMetadata.create(context, metadataConfig, basePath, config.getSpillableDir()));
() -> HoodieTableMetadata.create(context, metadataConfig, basePath, config.getSpillableDir(), true));
}

/**
Expand Down
35 changes: 25 additions & 10 deletions hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,9 @@ public class HoodieHFileReader<R extends IndexedRecord> implements HoodieFileRea
private Configuration conf;
private HFile.Reader reader;
private Schema schema;
// Scanner used to read individual keys. This is cached to prevent the overhead of opening the scanner for each
// key retrieval.
private HFileScanner keyScanner;

public static final String KEY_SCHEMA = "schema";
public static final String KEY_BLOOM_FILTER_META_BLOCK = "bloomFilter";
Expand Down Expand Up @@ -140,7 +143,7 @@ public Set<String> filterRowKeys(Set candidateRowKeys) {
public List<Pair<String, R>> readAllRecords(Schema writerSchema, Schema readerSchema) throws IOException {
List<Pair<String, R>> recordList = new LinkedList<>();
try {
HFileScanner scanner = reader.getScanner(false, false);
final HFileScanner scanner = reader.getScanner(false, false);
if (scanner.seekTo()) {
do {
Cell c = scanner.getKeyValue();
Expand Down Expand Up @@ -174,7 +177,7 @@ public boolean hasNext() {
// To handle when hasNext() is called multiple times for idempotency and/or the first time
if (this.next == null && !this.eof) {
if (!scanner.isSeeked() && scanner.seekTo()) {
this.next = (R)getRecordFromCell(scanner.getKeyValue(), getSchema(), readerSchema);
this.next = getRecordFromCell(scanner.getKeyValue(), getSchema(), readerSchema);
}
}
return this.next != null;
Expand All @@ -194,7 +197,7 @@ public R next() {
}
R retVal = this.next;
if (scanner.next()) {
this.next = (R)getRecordFromCell(scanner.getKeyValue(), getSchema(), readerSchema);
this.next = getRecordFromCell(scanner.getKeyValue(), getSchema(), readerSchema);
} else {
this.next = null;
this.eof = true;
Expand All @@ -209,12 +212,23 @@ public R next() {

@Override
public Option getRecordByKey(String key, Schema readerSchema) throws IOException {
HFileScanner scanner = reader.getScanner(false, true);
byte[] value = null;
KeyValue kv = new KeyValue(key.getBytes(), null, null, null);
if (scanner.seekTo(kv) == 0) {
Cell c = scanner.getKeyValue();
byte[] keyBytes = Arrays.copyOfRange(c.getRowArray(), c.getRowOffset(), c.getRowOffset() + c.getRowLength());
R record = getRecordFromCell(c, getSchema(), readerSchema);

synchronized (this) {
Comment thread
nsivabalan marked this conversation as resolved.
Outdated
if (keyScanner == null) {
keyScanner = reader.getScanner(true, true);
Comment thread
nsivabalan marked this conversation as resolved.
Outdated
}

if (keyScanner.seekTo(kv) == 0) {
Cell c = keyScanner.getKeyValue();
// Extract the byte value before releasing the lock since we cannot hold on to the returned cell afterwards
value = Arrays.copyOfRange(c.getValueArray(), c.getValueOffset(), c.getValueOffset() + c.getValueLength());
}
}

if (value != null) {
R record = (R)HoodieAvroUtils.bytesToAvro(value, getSchema(), readerSchema);
return Option.of(record);
}

Expand All @@ -232,12 +246,13 @@ public long getTotalRecords() {
}

@Override
public void close() {
public synchronized void close() {
try {
reader.close();
reader = null;
keyScanner = null;
} catch (IOException e) {
e.printStackTrace();
Comment thread
prashantwason marked this conversation as resolved.
throw new HoodieIOException("Error closing the hfile reader", e);
}
}

Expand Down
Loading