[HUDI-1532] Fixed suboptimal implementation of a magic sequence search

[vburenin]

What is the purpose of the pull request

Fixed suboptimal implementation of a magic sequence search that may take days on the file sizes of a few megabytes.
Instead of using 6 bytes buffer to find a magic sequence it uses a lot larger buffer that speeds up process like 170k times in some cases. The inefficiency is very noticeable when GCS or S3 storages are begin used.

Brief change log

Rewrote scanForNextAvailableBlockOffset function to use a large buffer size.

Verify this pull request

This pull request is already covered by existing tests

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

[vinothchandar]

@n3nash can you please review this?

[vinothchandar]

we buffer the underlying reads, correct? why is this happening?

[vburenin]

@vinothchandar I suspect the buffering of underlaying reads is on FileSystem driver, isn't it? If it is, GCS clearly not buffering it that can be seen in a form of a significant time distance (60 ms) between the calls to the readyFully method.

404120 [Executor task launch worker for task 268] INFO  org.apache.hudi.common.table.log.HoodieLogFileReader  - Current magic position: 263
404183 [Executor task launch worker for task 268] INFO  org.apache.hudi.common.table.log.HoodieLogFileReader  - Current magic position: 264
404246 [Executor task launch worker for task 268] INFO  org.apache.hudi.common.table.log.HoodieLogFileReader  - Current magic position: 265

[vinothchandar]

No I was referring to the BufferedFSInputStream code here

public HoodieLogFileReader(FileSystem fs, HoodieLogFile logFile, Schema readerSchema, int bufferSize,
      boolean readBlockLazily, boolean reverseReader) throws IOException {
    FSDataInputStream fsDataInputStream = fs.open(logFile.getPath(), bufferSize);
    if (fsDataInputStream.getWrappedStream() instanceof FSInputStream) {
      this.inputStream = new TimedFSDataInputStream(logFile.getPath(), new FSDataInputStream(
          new BufferedFSInputStream((FSInputStream) fsDataInputStream.getWrappedStream(), bufferSize)));
    } else {
      // fsDataInputStream.getWrappedStream() maybe a BufferedFSInputStream
      // need to wrap in another BufferedFSInputStream the make bufferSize work?
      this.inputStream = fsDataInputStream;
    }

    this.logFile = logFile;
    this.readerSchema = readerSchema;
    this.readBlockLazily = readBlockLazily;
    this.reverseReader = reverseReader;
    if (this.reverseReader) {
      this.reverseLogFilePosition = this.lastReverseLogFilePosition = fs.getFileStatus(logFile.getPath()).getLen();
    }
    addShutDownHook();
  }

[vburenin]

According to this stacktrace it is not the case:

	at com.google.cloud.hadoop.repackaged.gcs.com.google.api.services.storage.Storage$Objects$Get.executeMedia(Storage.java:6981)
	at com.google.cloud.hadoop.repackaged.gcs.com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadChannel.openStream(GoogleCloudStorageReadChannel.java:967)
	at com.google.cloud.hadoop.repackaged.gcs.com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadChannel.openContentChannel(GoogleCloudStorageReadChannel.java:772)
	at com.google.cloud.hadoop.repackaged.gcs.com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadChannel.performLazySeek(GoogleCloudStorageReadChannel.java:763)
	at com.google.cloud.hadoop.repackaged.gcs.com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadChannel.read(GoogleCloudStorageReadChannel.java:365)
	at com.google.cloud.hadoop.fs.gcs.GoogleHadoopFSInputStream.read(GoogleHadoopFSInputStream.java:131)
	- locked <0x0000000616319fb8> (a com.google.cloud.hadoop.fs.gcs.GoogleHadoopFSInputStream)
	at java.io.DataInputStream.read(DataInputStream.java:149)
	at java.io.DataInputStream.readFully(DataInputStream.java:195)
	at org.apache.hudi.common.table.log.HoodieLogFileReader.hasNextMagic(HoodieLogFileReader.java:339)
	at org.apache.hudi.common.table.log.HoodieLogFileReader.scanForNextAvailableBlockOffset(HoodieLogFileReader.java:280)```

[vinothchandar]

class GoogleHadoopFSInputStream extends FSInputStream { - Interesting.

Could you try printing out the actual class names with something like this? We can see if we can make it wrap properly

public HoodieLogFileReader(FileSystem fs, HoodieLogFile logFile, Schema readerSchema, int bufferSize,
      boolean readBlockLazily, boolean reverseReader) throws IOException {
    FSDataInputStream fsDataInputStream = fs.open(logFile.getPath(), bufferSize);

    System.err.println(">>>" + fsDataInputStream.getClass().getCanonicalName() + "," + 
        fsDataInputStream.getWrappedStream().getClass().getCanonicalName());

    if (fsDataInputStream.getWrappedStream() instanceof FSInputStream) {
      this.inputStream = new TimedFSDataInputStream(logFile.getPath(), new FSDataInputStream(
          new BufferedFSInputStream((FSInputStream) fsDataInputStream.getWrappedStream(), bufferSize)));
    } else {
      // fsDataInputStream.getWrappedStream() maybe a BufferedFSInputStream
      // need to wrap in another BufferedFSInputStream the make bufferSize work?
      this.inputStream = fsDataInputStream;
    }

[vburenin]

In process of building and trying.

The original search method is still O(m*n), which is also worth to optimize.

[vinothchandar]

Ack.

[codecov-io]

Codecov Report

Merging #2440 (e9de8a2) into master (e926c1a) will decrease coverage by 0.02%.
The diff coverage is 43.75%.

@@             Coverage Diff              @@
##             master    #2440      +/-   ##
============================================
- Coverage     50.73%   50.70%   -0.03%     
+ Complexity     3064     3059       -5     
============================================
  Files           419      419              
  Lines         18797    18810      +13     
  Branches       1922     1924       +2     
============================================
+ Hits           9536     9537       +1     
- Misses         8485     8495      +10     
- Partials        776      778       +2     

Flag	Coverage Δ	Complexity Δ
hudicli	37.26% <ø> (ø)	0.00 <ø> (ø)
hudiclient	100.00% <ø> (ø)	0.00 <ø> (ø)
hudicommon	51.96% <43.75%> (-0.06%)	0.00 <1.00> (ø)
hudiflink	10.20% <ø> (ø)	0.00 <ø> (ø)
hudihadoopmr	33.06% <ø> (ø)	0.00 <ø> (ø)
hudisparkdatasource	65.90% <ø> (ø)	0.00 <ø> (ø)
hudisync	48.61% <ø> (ø)	0.00 <ø> (ø)
huditimelineservice	66.84% <ø> (ø)	0.00 <ø> (ø)
hudiutilities	69.43% <ø> (ø)	0.00 <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ	Complexity Δ
...c/main/java/org/apache/hudi/common/fs/FSUtils.java	51.88% <ø> (ø)	61.00 <0.00> (ø)
...che/hudi/common/table/log/HoodieLogFileReader.java	67.85% <43.75%> (-3.22%)	22.00 <1.00> (-1.00)
.../apache/hudi/common/config/SerializableSchema.java	54.54% <0.00%> (-3.35%)	6.00% <0.00%> (ø%)
...ain/java/org/apache/hudi/avro/HoodieAvroUtils.java	56.09% <0.00%> (-1.19%)	37.00% <0.00%> (-4.00%)
...rg/apache/hudi/metadata/HoodieMetadataPayload.java	0.00% <0.00%> (ø)	0.00% <0.00%> (ø%)
...e/hudi/common/table/log/HoodieLogFormatWriter.java	79.68% <0.00%> (+1.56%)	26.00% <0.00%> (ø%)

[vburenin]

LOG.info("Class Name: " + fsDataInputStream.getWrappedStream().getClass().getName());

473840 [Executor task launch worker for task 267] INFO  org.apache.hudi.common.table.log.HoodieLogFileReader  - Class Name: org.apache.hadoop.fs.FSDataInputStream

FSDataInputStream has nothing to do with FSInputStream which in turn, makes it always non buffered.

[n3nash]

Instead of this, can we do the following in the constructor ?

    if (fsDataInputStream.getWrappedStream() instanceof FSInputStream ) {
      this.inputStream = new TimedFSDataInputStream(logFile.getPath(), new FSDataInputStream(
          new BufferedFSInputStream((FSInputStream) fsDataInputStream.getWrappedStream(), bufferSize)));
    } else if (**fsDataInputStream.getWrappedStream() instanceof FSDataInputStream**) {
 <initialize buffered input stream>
}

[n3nash]

After this change, we can leave the scanForNextAvailableBlockOffset() unchanged ? Also, please add comments as to why we need to add the extra check to make the inputStream Buffered

[vburenin]

I would consider keeping a new one as it does a lot less of memory copies and a lot less additional overhead, technically it is more efficient from the resource utilization stand point.

[n3nash]

@vburenin Just to confirm, by "lot less additional overhead" you refer to the in-memory to in-memory bytes copy operation that needs to be done for every 6 bytes vs 1MB in a single go in this implementation ? (Since the number of comparison of byte arrays is the same as looks like here -> https://github.com/apache/hbase/blob/master/hbase-common/src/main/java/org/apache/hadoop/hbase/util/Bytes.java#L2303). Can we quantify this overhead ?
Additionally, what is the reasoning behind keeping it at 1MB ?

[vburenin]

Buffered reader needs to check a few things to copy the right data, readFully logic itself is not trivial, there is also position modification each time it reads 6 bytes, etc so even without profiling it I bet the overhead is significant.
1MB seems like a good number to me, not too much, not too little. From my past experience dealing with FS IO going with blocks larger than 1MB was giving a diminished return. However, the best number would be the one that matches underlying block read size, but that depends on the reader which can be any.

[n3nash]

@vburenin According to the current code, we still are using BufferedReader for all cases except GCS, so that doesn't go away with this code in a generic way. Additionally, we need buffered reader code (the one I pointed above) anyways for GCS in the happy code path (without the need to find the magic header in corrupt blocks) since this method is only called when it encounters a corrupt block.

1. readFully does a bunch of if conditions so branching could cause some perf degradation here, don't see any other extra logic apart from copying bytes which is the same for irrespective of doing 6 bytes vs 1MB
2. Position modification in BufferedInputStream is a variable assignment which should not cause any overhead.

Agree with you that the best number should be the one that matches the underlying block size. It would be great if you can do some microbenchmarking here. I'm OK to land this once you can add the if check for BufferedInputStream since that is needed anyways ?

[vburenin]

All that simple logic runs every time for every byte offset, it all adds up no matter what, I also forgot to mention that we iterate over the data changing position just for 1 byte to copy next 6 bytes every time, so that it can potentially copy 6 times more data, which is not ideal.

I will add a BufferedInputStream as soon as I get back to work on Tuesday. It is super hard to find even 5 minutes when all kids are home.

[n3nash]

@vburenin Left a comment to restructure the code to support buffering, are you going to look into improving the O(m*n) search ?

[vburenin]

@vburenin Left a comment to restructure the code to support buffering, are you going to look into improving the O(m*n) search ?

At this point of time I think it is not necessary as the search pattern is trivial and the actually complexity is closer to O(n), the slowest point in the original code is memory copies and additional overhead associated with it.

[vinothchandar]

@n3nash can we take a call on this and get it into the current release. marking as blocker for now.

[vinothchandar]

@vburenin do you mind creating a JIRA for this issue.? We can give you perms if you can ping us your id from issue.apache.org/jira

[n3nash]

@vburenin any chance you can take a look at this soon ? I'd like to get this into 0.7.0

[vburenin]

@vburenin any chance you can take a look at this soon ? I'd like to get this into 0.7.0

I try to respond as soon as I can, but it is super hard to do it over the off days.

[vinothchandar]

@n3nash can you please drive this yourself if possible. we can get this into 0.7.0 if we can land tonight/early tomorrow

[vburenin]

Found 30 minutes to add a buffered reader. The addition doesn't look elegant though.

[n3nash]

@vburenin Thanks for the quick turnaround. I took the liberty to make a couple of changes, if it looks good to you, we can land this. I will timeout on this in the evening.

[vburenin]

Changes look good to me. Not sure why I didn't move 1MB into constants block though, it is so obvious.

[n3nash]

@vburenin That's fine, thanks for this PR. Like we discussed, let's do another audit of this code for performance issues when you get a chance next week. I will land this once the build succeeds.
@vinothchandar I'll be away till evening, please feel free to land if you notice the build succeed if I'm not around.