diff --git a/.github/workflows/ci-hadoop3.yml b/.github/workflows/ci-hadoop3.yml index c3304df13b..e111f869aa 100644 --- a/.github/workflows/ci-hadoop3.yml +++ b/.github/workflows/ci-hadoop3.yml @@ -31,7 +31,7 @@ jobs: strategy: fail-fast: false matrix: - java: [ { setup: '11', maven: '11' }, { setup: '17', maven: '17' } ] + java: [ { setup: '17', maven: '17' } ] name: Build Parquet with JDK ${{ matrix.java.setup }} steps: @@ -48,14 +48,10 @@ jobs: bash dev/ci-before_install.sh - name: install run: | - EXTRA_JAVA_TEST_ARGS=$(./mvnw help:evaluate -Dexpression=extraJavaTestArgs -q -DforceStdout) - export MAVEN_OPTS="$MAVEN_OPTS $EXTRA_JAVA_TEST_ARGS" ./mvnw install --batch-mode -DskipTests=true -Dmaven.javadoc.skip=true -Dsource.skip=true -Djava.version=${{ matrix.java.maven }} - name: verify env: TEST_CODECS: 'uncompressed,brotli,gzip,snappy,zstd' JAVA_VERSION: ${{ matrix.java.setup }} run: | - EXTRA_JAVA_TEST_ARGS=$(./mvnw help:evaluate -Dexpression=extraJavaTestArgs -q -DforceStdout) - export MAVEN_OPTS="$MAVEN_OPTS $EXTRA_JAVA_TEST_ARGS" ./mvnw verify --batch-mode javadoc:javadoc diff --git a/.github/workflows/vector-plugins.yml b/.github/workflows/vector-plugins.yml index bfa9c477a3..422a7efe1a 100644 --- a/.github/workflows/vector-plugins.yml +++ b/.github/workflows/vector-plugins.yml @@ -49,8 +49,6 @@ jobs: bash dev/ci-before_install.sh - name: install run: | - EXTRA_JAVA_TEST_ARGS=$(./mvnw help:evaluate -Dexpression=extraJavaTestArgs -q -DforceStdout) - export MAVEN_OPTS="$MAVEN_OPTS $EXTRA_JAVA_TEST_ARGS" # Spotless check uses palantir-java-format which relies on internal javac APIs # that are not available on all JDK versions (e.g. JDK 25+). Since the formatting # result is JDK-independent, running the check on JDK 17 alone is sufficient. @@ -64,8 +62,6 @@ jobs: TEST_CODECS: ${{ matrix.codes }} JAVA_VERSION: ${{ matrix.java }} run: | - EXTRA_JAVA_TEST_ARGS=$(./mvnw help:evaluate -Dexpression=extraJavaTestArgs -q -DforceStdout) - export MAVEN_OPTS="$MAVEN_OPTS $EXTRA_JAVA_TEST_ARGS" # Spotless check uses palantir-java-format which relies on internal javac APIs # that are not available on all JDK versions (e.g. JDK 25+). Since the formatting # result is JDK-independent, running the check on JDK 17 alone is sufficient. diff --git a/.gitignore b/.gitignore index c02d0f2222..f0d310a4a7 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,5 @@ mvn_install.log .vscode/* .DS_Store +.sdkmanrc + diff --git a/README.md b/README.md index 1ce03d5fd0..aef3cbad90 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ You can find additional details about the format and intended use cases in our [ ## Building -Parquet-Java uses Maven to build and depends on the thrift compiler (protoc is now managed by maven plugin). +Parquet-Java requires Java 17 or higher and uses Maven to build. It also depends on the thrift compiler (protoc is now managed by maven plugin). ### Install Thrift diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java index 50c2e344e2..84337635a5 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java @@ -778,7 +778,7 @@ public EncodingStats convertEncodingStats(List stats) { switch (stat.getPage_type()) { case DATA_PAGE_V2: builder.withV2Pages(); - // falls through + // falls through case DATA_PAGE: builder.addDataEncoding(getEncoding(stat.getEncoding()), stat.getCount()); break; diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java index c9391201f4..9024954bb9 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java @@ -456,7 +456,7 @@ private String cacheKey(CompressionCodecName codecName) { level = conf.get("parquet.compression.codec.zstd.level"); break; default: - // compression level is not supported; ignore it + // compression level is not supported; ignore it } String codecClass = codecName.getHadoopCompressionCodecClassName(); return level == null ? codecClass : codecClass + ":" + level; diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnIndexValidator.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnIndexValidator.java index 92f7db413c..5142506d2a 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnIndexValidator.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnIndexValidator.java @@ -546,7 +546,7 @@ private void validateBoundaryOrder( prevMaxValue::toString); break; case UNORDERED: - // No checks necessary. + // No checks necessary. } } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java index 312ccedf8f..4788c5f040 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java @@ -103,8 +103,8 @@ protected BytesCompressor createCompressor(final CompressionCodecName codecName) return new SnappyCompressor(); case ZSTD: return new ZstdCompressor(); - // todo: create class similar to the SnappyCompressor for zlib and exclude it as - // snappy is above since it also generates allocateDirect calls. + // todo: create class similar to the SnappyCompressor for zlib and exclude it as + // snappy is above since it also generates allocateDirect calls. default: return super.createCompressor(codecName); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index 9af4b4ac60..1f9e8d069c 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -65,6 +65,7 @@ import org.apache.parquet.HadoopReadOptions; import org.apache.parquet.ParquetReadOptions; import org.apache.parquet.Preconditions; +import org.apache.parquet.bytes.ByteBufferAllocator; import org.apache.parquet.bytes.ByteBufferInputStream; import org.apache.parquet.bytes.ByteBufferReleaser; import org.apache.parquet.bytes.BytesInput; @@ -1377,12 +1378,42 @@ private void readVectored(List allParts, ChunkListBuilder b totalSize += len; } LOG.debug("Reading {} bytes of data with vectored IO in {} ranges", totalSize, ranges.size()); - // Request a vectored read; - f.readVectored(ranges, options.getAllocator()); - int k = 0; - for (ConsecutivePartList consecutivePart : allParts) { - ParquetFileRange currRange = ranges.get(k++); - consecutivePart.readFromVectoredRange(currRange, builder); + // Use a capturing allocator to track all buffers allocated by Hadoop during vectored reads. + // The buffer returned from the read future may differ from the one originally allocated + // (e.g., ChecksumFileSystem wraps/copies buffers), so we must track the actual allocations. + List allocatedBuffers = new ArrayList<>(); + ByteBufferAllocator capturingAllocator = new ByteBufferAllocator() { + @Override + public ByteBuffer allocate(int size) { + ByteBuffer buf = options.getAllocator().allocate(size); + allocatedBuffers.add(buf); + return buf; + } + + @Override + public void release(ByteBuffer b) { + // Use identity comparison; ByteBuffer.equals() is content-based and could match wrong buffer + allocatedBuffers.removeIf(buf -> buf == b); + options.getAllocator().release(b); + } + + @Override + public boolean isDirect() { + return options.getAllocator().isDirect(); + } + }; + try { + // Request a vectored read; + f.readVectored(ranges, capturingAllocator); + int k = 0; + for (ConsecutivePartList consecutivePart : allParts) { + ParquetFileRange currRange = ranges.get(k++); + consecutivePart.readFromVectoredRange(currRange, builder); + } + } finally { + // Register all buffers allocated during vectored reads for release. + // In a finally block so buffers are not leaked on read failures. + builder.addBuffersToRelease(allocatedBuffers); } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CleanUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CleanUtil.java index 3cf88db874..fe139d47e0 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CleanUtil.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CleanUtil.java @@ -26,11 +26,8 @@ import org.slf4j.LoggerFactory; /** - * A helper class that uses {@code sun.misc.Unsafe#invokeCleaner(ByteBuffer)} - * reflectively to clean up direct buffers. - *

- * Strongly inspired by: - * Tomcat ByteBufferUtils + * A helper class which uses {@code sun.misc.Unsafe.invokeCleaner} to explicitly free + * direct ByteBuffers. */ public class CleanUtil { private static final Logger logger = LoggerFactory.getLogger(CleanUtil.class); diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java index e4be00947a..5a685d233f 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java @@ -34,6 +34,7 @@ import java.util.Set; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.io.compress.CompressionCodec; +import org.apache.hadoop.io.compress.zlib.ZlibCompressor; import org.apache.parquet.bytes.ByteBufferAllocator; import org.apache.parquet.bytes.ByteBufferReleaser; import org.apache.parquet.bytes.BytesInput; @@ -255,11 +256,14 @@ public org.apache.hadoop.io.compress.CompressionCodec getCodec(CompressionCodecN @Test public void cachingKeysGzip() { + // Hadoop's "zlib.compress.level" is read as a ZlibCompressor.CompressionLevel enum name + // (see ZlibFactory#getCompressionLevel), so it must hold an enum constant name rather than a + // raw numeric value. Hadoop 3.4.x's BuiltInGzipCompressor enforces this, unlike older versions. Configuration config_zlib_2 = new Configuration(); - config_zlib_2.set("zlib.compress.level", "2"); + config_zlib_2.set("zlib.compress.level", ZlibCompressor.CompressionLevel.TWO.name()); Configuration config_zlib_5 = new Configuration(); - config_zlib_5.set("zlib.compress.level", "5"); + config_zlib_5.set("zlib.compress.level", ZlibCompressor.CompressionLevel.FIVE.name()); final CodecFactory codecFactory_2 = new PublicCodecFactory(config_zlib_2, pageSize); final CodecFactory codecFactory_5 = new PublicCodecFactory(config_zlib_5, pageSize); diff --git a/parquet-plugins/parquet-encoding-vector/pom.xml b/parquet-plugins/parquet-encoding-vector/pom.xml index 06c4084be1..4ceb0b43d4 100644 --- a/parquet-plugins/parquet-encoding-vector/pom.xml +++ b/parquet-plugins/parquet-encoding-vector/pom.xml @@ -34,11 +34,6 @@ Apache Parquet Encodings Vector https://parquet.apache.org - - - 17 - - org.apache.parquet diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/BufferedProtocolReadToWrite.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/BufferedProtocolReadToWrite.java index e1c2aae57a..07c31c358c 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/BufferedProtocolReadToWrite.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/BufferedProtocolReadToWrite.java @@ -226,7 +226,7 @@ private boolean readOneValue(TProtocol in, byte type, List buffer, Thrif writeShortAction(buffer, s); break; case TType.ENUM: // same as i32 => actually never seen in the protocol layer as enums are written as a i32 - // field + // field case TType.I32: final int i = in.readI32(); checkEnum(expectedType, i); diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ProtocolReadToWrite.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ProtocolReadToWrite.java index 94c68b0081..830a5c605b 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ProtocolReadToWrite.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ProtocolReadToWrite.java @@ -74,7 +74,7 @@ void readOneValue(TProtocol in, TProtocol out, byte type) throws TException { out.writeI16(in.readI16()); break; case TType.ENUM: // same as i32 => actually never seen in the protocol layer as enums are written as a i32 - // field + // field case TType.I32: out.writeI32(in.readI32()); break; diff --git a/pom.xml b/pom.xml index 62bdb32623..acf16decfd 100644 --- a/pom.xml +++ b/pom.xml @@ -69,7 +69,7 @@ - 11 + 17 github com.fasterxml.jackson.core com.fasterxml.jackson.datatype @@ -80,10 +80,10 @@ 2.22 0.26.1 1.3.2 - 2.46.1 + 3.5.1 shaded.parquet - 3.3.0 + 3.4.3 2.13.0 1.17.0 thrift @@ -122,23 +122,6 @@ true - - - -XX:+IgnoreUnrecognizedVMOptions - --add-opens=java.base/java.lang=ALL-UNNAMED - --add-opens=java.base/java.lang.invoke=ALL-UNNAMED - --add-opens=java.base/java.lang.reflect=ALL-UNNAMED - --add-opens=java.base/java.io=ALL-UNNAMED - --add-opens=java.base/java.net=ALL-UNNAMED - --add-opens=java.base/java.nio=ALL-UNNAMED - --add-opens=java.base/java.util=ALL-UNNAMED - --add-opens=java.base/java.util.concurrent=ALL-UNNAMED - --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED - --add-opens=java.base/sun.nio.ch=ALL-UNNAMED - --add-opens=java.base/sun.nio.cs=ALL-UNNAMED - --add-opens=java.base/sun.security.action=ALL-UNNAMED - --add-opens=java.base/sun.util.calendar=ALL-UNNAMED - @@ -299,6 +282,21 @@ maven-enforcer-plugin + + enforce-java-version + + enforce + + + + + [17,) + Java 17 or higher is required to build parquet-java. + + + true + + enforce-banned-dependencies @@ -447,9 +445,6 @@ org.apache.maven.plugins maven-failsafe-plugin - - ${extraJavaTestArgs} - @@ -464,7 +459,7 @@ org.apache.maven.plugins maven-surefire-plugin - ${surefire.argLine} ${extraJavaTestArgs} + ${surefire.argLine} + org.apache.parquet.internal.column.columnindex.IndexIterator org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesReader#gatherElementDataFromStreams(byte[])