[format] Close the Avro stats extractor's stream on a corrupt file

[LuciferYang]

Purpose

close #9572

AvroSimpleStatsExtractor.extractWithFileInfo opens the file and hands the stream to getRowCount, whose try-with-resources binds the DataFileStream rather than the stream:

SeekableInputStream fileInputStream = fileIO.newInputStream(path);
long rowCount = getRowCount(fileInputStream);

When the DataFileStream constructor throws, the resource was never bound and no local holds the input, so nothing closes it. Three ways it throws: a non-Avro file fails the magic check, an empty or truncated file hits EOF in readMagic, and an unrecognised avro.codec makes CodecFactory.fromString throw. This closes the stream on the failure path, the way AvroBulkFormat.createReaderFromPath in the same package already does.

Two choices worth calling out. The catch is on Throwable, mirroring AvroBulkFormat in the same package; it has to be at least RuntimeException, because the codec case is unchecked and catch (IOException) would miss it. And closing only on failure rather than wrapping the method in try-with-resources, so the success path still closes exactly once: DataFileStream.close closes the stream it was given, and an outer try-with-resources would close it a second time. The implementations I looked at are idempotent, so a double close is harmless today, but it depends on that and the tests below pin the single close instead.

Only migrate and clone reach this extractor with a file Paimon did not write; both write paths build stats from the collector for avro (RollingFileWriter.createStatsProducer, KeyValueFileWriterFactory.statsProducer). One attempt leaks one descriptor. What makes it worth fixing is that the corrupt-file path is not exotic: a zero-byte file left by a failed task is an ordinary thing to find in a Hive table directory, and the migrate scan only skips names beginning with _ or ..

Tests

AvroSimpleStatsExtractorLeakTest wraps LocalFileIO so every stream counts its own close() calls, and asserts exactly one close in all four cases rather than just "was closed": a non-Avro file, an empty file, an unrecognised codec, and a valid three-row file. The codec case takes a real zstd file and renames zstandard to zstandarX in the header, keeping the length so the header still parses and the failure comes from CodecFactory.fromString; it asserts AvroRuntimeException with that message, since that case is the whole reason the catch is on Throwable.

The three corrupt cases fail against the unfixed extractor. The valid-file case passes there too, which is the point of it: it guards the single close, which is what a return to try-with-resources would break.

mvn -pl paimon-format test on JDK 8: 600 tests, 0 failures. spotless:check and checkstyle:check are clean.

[JingsongLi]

+1

[LuciferYang]

Thank you @JingsongLi