You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2021/10/20 01:14:33 UTC
[GitHub] [hudi] vinothchandar commented on a change in pull request #3762: [HUDI-1294] Adding inline read and seek based read(batch get) for hfile log blocks in metadata table
vinothchandar commented on a change in pull request #3762:
URL: https://github.com/apache/hudi/pull/3762#discussion_r732338819
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
##########
@@ -115,6 +115,18 @@
.sinceVersion("0.7.0")
.withDocumentation("Parallelism to use, when listing the table on lake storage.");
+ public static final ConfigProperty<Boolean> ENABLE_INLINE_READING_LOG_FILES = ConfigProperty
+ .key(METADATA_PREFIX + ".enable.inline.reading.log.files")
+ .defaultValue(true)
+ .sinceVersion("0.10.0")
+ .withDocumentation("Enable inline reading of Log files");
+
+ public static final ConfigProperty<Boolean> ENABLE_FULL_SCAN_LOG_FILES = ConfigProperty
+ .key(METADATA_PREFIX + ".enable.full.scan.log.files")
+ .defaultValue(true)
+ .sinceVersion("0.10.0")
+ .withDocumentation("Enable full scanning of log files while reading log records");
Review comment:
little bit more context?
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
##########
@@ -115,6 +115,18 @@
.sinceVersion("0.7.0")
.withDocumentation("Parallelism to use, when listing the table on lake storage.");
+ public static final ConfigProperty<Boolean> ENABLE_INLINE_READING_LOG_FILES = ConfigProperty
+ .key(METADATA_PREFIX + ".enable.inline.reading.log.files")
Review comment:
just `enable.inline.reading`
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
##########
@@ -141,6 +148,50 @@ public HoodieLogBlockType getBlockType() {
return baos.toByteArray();
}
+ @Override
+ protected void createRecordsFromContentBytes() throws IOException {
+ if (enableInlineReading) {
+ getRecords(Collections.emptyList());
+ } else {
+ super.createRecordsFromContentBytes();
+ }
+ }
+
+ @Override
+ public List<IndexedRecord> getRecords(List<String> keys) throws IOException {
+ readWithInlineFS(keys);
+ return records;
+ }
+
+ private void readWithInlineFS(List<String> keys) throws IOException {
+ boolean enableFullScan = keys.isEmpty();
+ // Get schema from the header
+ Schema writerSchema = new Schema.Parser().parse(super.getLogBlockHeader().get(HeaderMetadataType.SCHEMA));
+ // If readerSchema was not present, use writerSchema
+ if (schema == null) {
+ schema = writerSchema;
+ }
+ Configuration conf = new Configuration();
Review comment:
could nt we get the config object from someplace else? if you just intialize it here, would nt we lose the configurations set at the higher layers (query engine or spark writer)
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java
##########
@@ -132,18 +149,31 @@ protected AbstractHoodieLogRecordScanner(FileSystem fs, String basePath, List<St
}
this.totalLogFiles.addAndGet(logFilePaths.size());
this.logFilePaths = logFilePaths;
- this.readBlocksLazily = readBlocksLazily;
this.reverseReader = reverseReader;
+ this.readBlocksLazily = readBlocksLazily;
this.fs = fs;
this.bufferSize = bufferSize;
this.instantRange = instantRange;
this.withOperationField = withOperationField;
+ this.enableInlineReading = enableInlineReading;
+ this.enableFullScan = enableFullScan;
+ if (!enableFullScan) {
+ ValidationUtils.checkArgument(enableInlineReading, "Inline should be enabled if full scan is not enabled");
+ }
}
- /**
- * Scan Log files.
- */
public void scan() {
+ scan(Collections.emptyList());
Review comment:
using empty list to scan all keys is kind of counter intuitive.
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -169,16 +181,98 @@ private void initIfNeeded() {
}
}
+ protected List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> getRecordsByKeysFromMetadata(List<String> keys, String partitionName) {
Review comment:
this method is too long.
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -169,16 +181,98 @@ private void initIfNeeded() {
}
}
+ protected List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> getRecordsByKeysFromMetadata(List<String> keys, String partitionName) {
+ Pair<HoodieFileReader, HoodieMetadataMergedLogRecordReader> readers = openReadersIfNeeded(keys.get(0), partitionName);
+ try {
+ List<Long> timings = new ArrayList<>();
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ HoodieFileReader baseFileReader = readers.getKey();
+ HoodieMetadataMergedLogRecordReader logRecordScanner = readers.getRight();
+
+ List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> result = new ArrayList<>();
+ // local map to assist in merging with base file records
+ Map<String, Option<HoodieRecord<HoodieMetadataPayload>>> logRecords = new HashMap<>();
+
+ // Retrieve records from log file
+ timer.startTimer();
Review comment:
IMO, we need some better ways of abstracting the two code paths for scannign and seeking
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java
##########
@@ -132,18 +149,31 @@ protected AbstractHoodieLogRecordScanner(FileSystem fs, String basePath, List<St
}
this.totalLogFiles.addAndGet(logFilePaths.size());
this.logFilePaths = logFilePaths;
- this.readBlocksLazily = readBlocksLazily;
this.reverseReader = reverseReader;
+ this.readBlocksLazily = readBlocksLazily;
this.fs = fs;
this.bufferSize = bufferSize;
this.instantRange = instantRange;
this.withOperationField = withOperationField;
+ this.enableInlineReading = enableInlineReading;
+ this.enableFullScan = enableFullScan;
+ if (!enableFullScan) {
+ ValidationUtils.checkArgument(enableInlineReading, "Inline should be enabled if full scan is not enabled");
+ }
}
- /**
- * Scan Log files.
- */
public void scan() {
+ scan(Collections.emptyList());
Review comment:
using empty list to scan all keys is kind of counter intuitive
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
##########
@@ -115,6 +115,18 @@
.sinceVersion("0.7.0")
.withDocumentation("Parallelism to use, when listing the table on lake storage.");
+ public static final ConfigProperty<Boolean> ENABLE_INLINE_READING_LOG_FILES = ConfigProperty
+ .key(METADATA_PREFIX + ".enable.inline.reading.log.files")
+ .defaultValue(true)
+ .sinceVersion("0.10.0")
+ .withDocumentation("Enable inline reading of Log files");
Review comment:
more descriptive docs?
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java
##########
@@ -132,18 +149,31 @@ protected AbstractHoodieLogRecordScanner(FileSystem fs, String basePath, List<St
}
this.totalLogFiles.addAndGet(logFilePaths.size());
this.logFilePaths = logFilePaths;
- this.readBlocksLazily = readBlocksLazily;
this.reverseReader = reverseReader;
+ this.readBlocksLazily = readBlocksLazily;
this.fs = fs;
this.bufferSize = bufferSize;
this.instantRange = instantRange;
this.withOperationField = withOperationField;
+ this.enableInlineReading = enableInlineReading;
Review comment:
why do we need the two flags? can't we assume full scan by default and then do inline reading as needed?
##########
File path: hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.bloom.BloomFilterFactory;
+import org.apache.hudi.common.bloom.BloomFilterTypeCode;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.util.Pair;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.hudi.common.testutils.FileSystemTestUtils.RANDOM;
+import static org.apache.hudi.common.testutils.SchemaTestUtil.getSchemaFromResource;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TestHoodieHFileReaderWriter {
+ private final Path filePath = new Path(System.getProperty("java.io.tmpdir") + "/f1_1-0-1_000.hfile");
Review comment:
can we use the junit temporary folder creation for the testing?
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
##########
@@ -141,6 +148,50 @@ public HoodieLogBlockType getBlockType() {
return baos.toByteArray();
}
+ @Override
+ protected void createRecordsFromContentBytes() throws IOException {
+ if (enableInlineReading) {
+ getRecords(Collections.emptyList());
+ } else {
+ super.createRecordsFromContentBytes();
+ }
+ }
+
+ @Override
+ public List<IndexedRecord> getRecords(List<String> keys) throws IOException {
+ readWithInlineFS(keys);
+ return records;
+ }
+
+ private void readWithInlineFS(List<String> keys) throws IOException {
+ boolean enableFullScan = keys.isEmpty();
+ // Get schema from the header
+ Schema writerSchema = new Schema.Parser().parse(super.getLogBlockHeader().get(HeaderMetadataType.SCHEMA));
+ // If readerSchema was not present, use writerSchema
+ if (schema == null) {
+ schema = writerSchema;
+ }
+ Configuration conf = new Configuration();
Review comment:
could nt we get the config object from someplace else? if you just intialize it here, would nt we lose the configurations set at the higher layers (query engine or spark writer)
##########
File path: hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileReader.java
##########
@@ -164,6 +174,25 @@ public BloomFilter readBloomFilter() {
return readAllRecords(schema, schema);
}
+ public List<Pair<String, R>> readRecordsByKey(List<String> keys) throws IOException {
Review comment:
just `readRecords()`
##########
File path: hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileReader.java
##########
@@ -250,6 +279,9 @@ public synchronized void close() {
try {
reader.close();
reader = null;
+ if (fsDataInputStream != null) {
Review comment:
this was a leak?
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java
##########
@@ -342,17 +377,17 @@ private void processDataBlock(HoodieDataBlock dataBlock) throws Exception {
/**
* Process the set of log blocks belonging to the last instant which is read fully.
*/
- private void processQueuedBlocksForInstant(Deque<HoodieLogBlock> lastBlocks, int numLogFilesSeen) throws Exception {
- while (!lastBlocks.isEmpty()) {
- LOG.info("Number of remaining logblocks to merge " + lastBlocks.size());
+ private void processQueuedBlocksForInstant(Deque<HoodieLogBlock> logBlocks, int numLogFilesSeen, List<String> keys) throws Exception {
Review comment:
i don't per se like how we are passing the keys up and down. but yeah can't think of better ways either.
##########
File path: hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java
##########
@@ -111,6 +111,17 @@ public static HoodieLogBlock getBlock(HoodieLogBlockType logDataBlockFormat, Lis
return records;
}
+ /**
+ * Batch get of keys of interest. Implementation can choose to either do full scan and return matched entries or
+ * do a seek based parsing and return matched entries.
+ * @param keys keys of interest.
+ * @return List of IndexedRecords for the keys of interest.
+ * @throws IOException
+ */
+ public List<IndexedRecord> getRecords(List<String> keys) throws IOException {
+ throw new UnsupportedOperationException("On demand batch get based on keys not supported");
Review comment:
key based fetch not allowed?
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
##########
@@ -200,8 +201,49 @@ protected BaseTableMetadata(HoodieEngineContext engineContext, HoodieMetadataCon
return statuses;
}
+ Map<String, FileStatus[]> fetchAllFilesInPartitionPaths(List<Path> partitionPaths) throws IOException {
+ Map<String, Path> partitionInfo = new HashMap<>();
+ boolean foundNonPartitionedPath = false;
+ for (Path partitionPath: partitionPaths) {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(dataBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ if (partitionInfo.size() > 1) {
+ throw new HoodieMetadataException("Found mix of partitioned and non partitioned paths while fetching data from metadata table");
+ }
+ partitionInfo.put(NON_PARTITIONED_NAME, partitionPath);
+ foundNonPartitionedPath = true;
+ } else {
+ if (foundNonPartitionedPath) {
+ throw new HoodieMetadataException("Found mix of partitioned and non partitioned paths while fetching data from metadata table");
+ }
+ partitionInfo.put(partitionName, partitionPath);
+ }
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> partitionsFileStatus =
+ getRecordsByKeysFromMetadata(new ArrayList<>(partitionInfo.keySet()), MetadataPartitionType.FILES.partitionPath());
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+ Map<String, FileStatus[]> result = new HashMap<>();
+
+ for (Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>> entry: partitionsFileStatus) {
+ if (entry.getValue().isPresent()) {
+ if (!entry.getValue().get().getData().getDeletions().isEmpty()) {
+ throw new HoodieMetadataException("Metadata record for partition " + entry.getKey() + " is inconsistent: "
+ + entry.getValue().get().getData());
+ }
+ result.put(partitionInfo.get(entry.getKey()).toString(), entry.getValue().get().getData().getFileStatuses(hadoopConf.get(), partitionInfo.get(entry.getKey())));
+ }
+ }
+
+ LOG.info("Listed files in partitions from metadata: partition list =" + Arrays.toString(partitionPaths.toArray()));
+ return result;
+ }
+
protected abstract Option<HoodieRecord<HoodieMetadataPayload>> getRecordByKeyFromMetadata(String key, String partitionName);
+ protected abstract List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> getRecordsByKeysFromMetadata(List<String> key, String partitionName);
Review comment:
drop the `FromMetadata` (here and the method above?)
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
##########
@@ -200,8 +201,49 @@ protected BaseTableMetadata(HoodieEngineContext engineContext, HoodieMetadataCon
return statuses;
}
+ Map<String, FileStatus[]> fetchAllFilesInPartitionPaths(List<Path> partitionPaths) throws IOException {
Review comment:
unit test this, if not exists?
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
##########
@@ -169,16 +181,98 @@ private void initIfNeeded() {
}
}
+ protected List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> getRecordsByKeysFromMetadata(List<String> keys, String partitionName) {
Review comment:
Does this code exist in another place for code reuse?
##########
File path: hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
##########
@@ -200,8 +201,49 @@ protected BaseTableMetadata(HoodieEngineContext engineContext, HoodieMetadataCon
return statuses;
}
+ Map<String, FileStatus[]> fetchAllFilesInPartitionPaths(List<Path> partitionPaths) throws IOException {
+ Map<String, Path> partitionInfo = new HashMap<>();
+ boolean foundNonPartitionedPath = false;
+ for (Path partitionPath: partitionPaths) {
+ String partitionName = FSUtils.getRelativePartitionPath(new Path(dataBasePath), partitionPath);
+ if (partitionName.isEmpty()) {
+ if (partitionInfo.size() > 1) {
+ throw new HoodieMetadataException("Found mix of partitioned and non partitioned paths while fetching data from metadata table");
+ }
+ partitionInfo.put(NON_PARTITIONED_NAME, partitionPath);
+ foundNonPartitionedPath = true;
+ } else {
+ if (foundNonPartitionedPath) {
+ throw new HoodieMetadataException("Found mix of partitioned and non partitioned paths while fetching data from metadata table");
+ }
+ partitionInfo.put(partitionName, partitionPath);
+ }
+ }
+
+ HoodieTimer timer = new HoodieTimer().startTimer();
+ List<Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>>> partitionsFileStatus =
+ getRecordsByKeysFromMetadata(new ArrayList<>(partitionInfo.keySet()), MetadataPartitionType.FILES.partitionPath());
+ metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
+ Map<String, FileStatus[]> result = new HashMap<>();
+
+ for (Pair<String, Option<HoodieRecord<HoodieMetadataPayload>>> entry: partitionsFileStatus) {
Review comment:
lets break this method down into more readable pieces
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org