Spark/Flink: ORC support estimated length for unclosed file.

[hililiwei]

This PR consists of the following parts:

• Port ORC rolling writer and unit test to Spark 3.1 3.0 2.4 and Flink 1.13 1.12, refer: #3784
• Clear the Formart that are no longer needed.

[openinx]

As this is a minor change for the specific engine version, could you please create a single PR to include all those changes for all engine versions ?

[hililiwei]

As this is a minor change for the specific engine version, could you please create a single PR to include all those changes for all engine versions ?

Is the modification of Flink also submitted here or a separate PR?

[openinx]

I think it's fair to make all the engine's changes into a single PR.

[hililiwei]

I think it's fair to make all the engine's changes into a single PR.

Updated code and PR description.

[openinx]

I just checked all the ORC rolling file writer unit tests, all unit tests seems to be enabled after this PR:

➜ iceberg git:(4419) find . -type f -name '*.java' | xargs grep -i 'Assume.*ORC'
./mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerWithEngine.java: assumeTrue(isVectorized && FileFormat.ORC.equals(fileFormat));
./hive3/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java: // reader will assume that the ORC file ends at the task's start + length, and might fail reading the tail..
./data/src/test/java/org/apache/iceberg/data/TestMetricsRowGroupFilter.java: Assume.assumeFalse("ORC row group filter does not support StringStartsWith", format == FileFormat.ORC);
./data/src/test/java/org/apache/iceberg/data/TestMetricsRowGroupFilter.java: Assume.assumeFalse("ORC row group filter does not support StringStartsWith", format == FileFormat.ORC);
./spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java: Assume.assumeFalse(fileFormat == FileFormat.ORC && vectorized);
./spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java: Assume.assumeFalse(fileFormat == FileFormat.ORC && vectorized);
./spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java: Assume.assumeFalse(fileFormat == FileFormat.ORC && vectorized);
➜ iceberg git:(4419) find . -type f -name '*.scala' | xargs grep -i 'TODO.*orc' ➜ iceberg git:(4419) find . -type f -name '*.java' | xargs grep -i 'Assume.*ORC'
./mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerWithEngine.java: assumeTrue(isVectorized && FileFormat.ORC.equals(fileFormat));
./hive3/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java: // reader will assume that the ORC file ends at the task's start + length, and might fail reading the tail..
./data/src/test/java/org/apache/iceberg/data/TestMetricsRowGroupFilter.java: Assume.assumeFalse("ORC row group filter does not support StringStartsWith", format == FileFormat.ORC);
./data/src/test/java/org/apache/iceberg/data/TestMetricsRowGroupFilter.java: Assume.assumeFalse("ORC row group filter does not support StringStartsWith", format == FileFormat.ORC);
./spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java: Assume.assumeFalse(fileFormat == FileFormat.ORC && vectorized);
./spark/v3.0/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java: Assume.assumeFalse(fileFormat == FileFormat.ORC && vectorized);
./spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java: Assume.assumeFalse(fileFormat == FileFormat.ORC && vectorized);

[openinx]

Looks pretty great to me !