[FLINK-33059] Support transparent compression for file-connector for all file input formats

[echauchot]

What is the purpose of the change

Support transparent compression for file-connector for all file input formats.

Brief change log

• Force reading the whole file split (like it is done for binary input formats) on compressed (unsplittable) files
• add FileInputFormatTest#testFileInputFormatWithCompressionFromFileSource
• generalize FileInputFormatTest#testFileInputFormatWithCompression to more that deflate format

Verifying this change

FileInputFormatTest#testFileInputFormatWithCompressionFromFileSource

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? website

[flinkbot]

CI report:

• 3ad7e33 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[echauchot]

Hi @rmetzger, I saw you authored parts of this code, can you please do a review or point me to another reviewer ?

[echauchot]

R: @xintongsong I see your name in that code history, would you have time to take a look ?

[echauchot]

@tzulitai you offered help for reviewing, don't hesitate to ping me on this when you have time.

[ferenc-csaky]

The added logic makes sense IMO.

Added 2 comments, but in general I would consider separating the whole INFLATER_INPUT_STREAM_FACTORIES into a new class as there are a couple functions that uses it and seems quite detachable from FlieInputFormats.

After a quick peek I think something like the following could work:

public class InflaterInputStreamFactories {

  public static void register(String fileExt, InflaterInputStreamFactory<?> factory) { ... }

  public static InflaterInputStreamFactory<?> get(Path path) { ... }

  private static InflaterInputStreamFactory<?> get(String fileExt) { ... }

  @VisibleForTesting
  public static Set<String> getSupportedCompressionFormats() { ... }
}

Also, ConcurrentHashMap can be utilized insead of the synchronized block, but other than that the current logic could be moved as is now.

This probably goes beyond the current PR, but I think it worth to note it. WDYT?

[ferenc-csaky]

I'd mark this @VisibleForTesting, because only tests use this function.

[echauchot]

👍 thx for pointing out

[ferenc-csaky]

I think instead of bringing the specific FileInputFormat into this general utility, it would be cleaner to pass Set<String> extensions as a parameter.

[echauchot]

Agree, better to be more generic here, thx !

[echauchot]

The added logic makes sense IMO.

Added 2 comments, but in general I would consider separating the whole INFLATER_INPUT_STREAM_FACTORIES into a new class as there are a couple functions that uses it and seems quite detachable from FlieInputFormats.

After a quick peek I think something like the following could work:

public class InflaterInputStreamFactories {

  public static void register(String fileExt, InflaterInputStreamFactory<?> factory) { ... }

  public static InflaterInputStreamFactory<?> get(Path path) { ... }

  private static InflaterInputStreamFactory<?> get(String fileExt) { ... }

  @VisibleForTesting
  public static Set<String> getSupportedCompressionFormats() { ... }
}

Also, ConcurrentHashMap can be utilized insead of the synchronized block, but other than that the current logic could be moved as is now.

This probably goes beyond the current PR, but I think it worth to note it. WDYT?

I agree with the ConcurrentHashMap suggestion. Regarding creating a class just for wrapping a map that is used only in the FileInputFormat it seems overkill to me. An anyway it is indeed outside of the scope of this filesize-fix PR.

[echauchot]

@ferenc-csaky thanks for reviewing this PR ! I have addressed your comments, do I have your LGTM when the tests pass ?

[ferenc-csaky]

LGTM, if tests pass! 👍

[ruslandanilin]

Thank you @echauchot !

[echauchot]

Thank you @echauchot !

My pleasure ! Merging