linliu-code commented on code in PR #9762:
URL: https://github.com/apache/hudi/pull/9762#discussion_r1333572375
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java:
##########
@@ -539,6 +542,121 @@ private void updateBlockSequenceTracker(HoodieLogBlock
logBlock, String instantT
}
}
+ private void scanInternalV3(Option<KeySpec> keySpecOption, boolean
skipProcessingBlocks) {
+ totalLogFiles = new AtomicLong(0);
+ totalRollbacks = new AtomicLong(0);
+ totalCorruptBlocks = new AtomicLong(0);
+ totalLogBlocks = new AtomicLong(0);
+ totalLogRecords = new AtomicLong(0);
+ ConcurrentMap<String, List<HoodieLogBlock>> instantToBlocksMap = new
ConcurrentHashMap<>();
+ ConcurrentMap<String, String> blockTimeToCompactionBlockTimeMap = new
ConcurrentHashMap<>();
+ currentInstantLogBlocks = new ConcurrentLinkedDeque<>();
+ Set<String> targetRollbackInstants = ConcurrentHashMap.newKeySet();
+ Set<HoodieLogFile> scannedLogFiles = ConcurrentHashMap.newKeySet();
+
+ try (HoodieLogFormatReader logFormatReaderWrapper = new
HoodieLogFormatReader(fs,
+ logFilePaths.stream().map(logFile -> new HoodieLogFile(new
CachingPath(logFile))).collect(Collectors.toList()),
+ readerSchema, true, reverseReader, bufferSize, shouldLookupRecords(),
recordKeyField, internalSchema)) {
+ while (logFormatReaderWrapper.hasNext()) {
+ HoodieLogFile logFile = logFormatReaderWrapper.getLogFile();
+ LOG.info("Scanning log file " + logFile);
+ scannedLogFiles.add(logFile);
+ totalLogFiles.set(scannedLogFiles.size());
+ HoodieLogBlock logBlock = logFormatReaderWrapper.next();
+
+ if (shouldSkipBlock(logBlock)) {
+ continue;
+ }
+
+ switch (logBlock.getBlockType()) {
+ case HFILE_DATA_BLOCK:
+ case AVRO_DATA_BLOCK:
+ case PARQUET_DATA_BLOCK:
+ case DELETE_BLOCK:
+ processRegularDataBlock(logBlock, instantToBlocksMap,
blockTimeToCompactionBlockTimeMap, currentInstantLogBlocks);
+ break;
+ case COMMAND_BLOCK:
+ processCommandBlock(logBlock, instantToBlocksMap,
targetRollbackInstants);
+ break;
+ default:
+ throw new UnsupportedOperationException("Block type not yet
supported.");
+ }
+ }
+ // merge the last read block when all the blocks are done reading
+ if (!currentInstantLogBlocks.isEmpty() && !skipProcessingBlocks) {
+ LOG.info("Merging the final data blocks");
+ processQueuedBlocksForInstant(currentInstantLogBlocks,
scannedLogFiles.size(), keySpecOption);
+ }
+ } catch (IOException 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);
+ throw new HoodieException("Exception when reading log file ", e);
+ }
+ }
+
+ private boolean shouldSkipBlock(HoodieLogBlock logBlock) {
+ if (logBlock.getBlockType().equals(CORRUPT_BLOCK)) {
+ totalCorruptBlocks.incrementAndGet();
+ return true;
+ }
+ HoodieTimeline commitsTimeline =
this.hoodieTableMetaClient.getCommitsTimeline();
+ HoodieTimeline completedInstantsTimeline =
commitsTimeline.filterCompletedInstants();
+ HoodieTimeline inflightInstantsTimeline =
commitsTimeline.filterInflights();
+ String instantTime = logBlock.getLogBlockHeader().get(INSTANT_TIME);
+ if (logBlock.getBlockType() != COMMAND_BLOCK) {
+ if
(!completedInstantsTimeline.containsOrBeforeTimelineStarts(instantTime)
+ || inflightInstantsTimeline.containsInstant(instantTime)) {
+ // hit an uncommitted block possibly from a failed write, move to the
next one and skip processing this one
+ return true;
+ }
+ if (instantRange.isPresent() &&
!instantRange.get().isInRange(instantTime)) {
+ // filter the log block by instant range
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void processRegularDataBlock(HoodieLogBlock logBlock,
ConcurrentMap<String, List<HoodieLogBlock>> instantToBlocksMap,
ConcurrentMap<String, String> blockTimeToCompactionBlockTimeMap,
+ Deque<HoodieLogBlock> processingDeque) {
Review Comment:
a format problem?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
