You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@parquet.apache.org by "gszadovszky (via GitHub)" <gi...@apache.org> on 2024/02/28 15:47:13 UTC
[PR] PARQUET-2437: Avoid flushing at Parquet writes after an exception [parquet-mr]
gszadovszky opened a new pull request, #1285:
URL: https://github.com/apache/parquet-mr/pull/1285
Make sure you have checked _all_ steps below.
### Jira
- [x] My PR addresses the following [PARQUET-2437](https://issues.apache.org/jira/browse/PARQUET-2437) issues and references
them in the PR title. For example, "PARQUET-1234: My Parquet PR"
- https://issues.apache.org/jira/browse/PARQUET-XXX
- In case you are adding a dependency, check if the license complies with
the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
### Tests
- [x] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [x] My commits all reference Jira issues in their subject lines. In addition, my commits follow the guidelines
from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
1. Subject is limited to 50 characters (not including Jira issue reference)
1. Subject does not end with a period
1. Subject uses the imperative mood ("add", not "adding")
1. Body wraps at 72 characters
1. Body explains "what" and "why", not "how"
### Style
- [x] My contribution adheres to the code style guidelines and Spotless passes.
- To apply the necessary changes, run `mvn spotless:apply -Pvector-plugins`
### Documentation
- [x] In case of new functionality, my PR adds documentation that describes how to use it.
- All the public functions and the classes in the PR contain Javadoc that explain what it does
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] PARQUET-2437: Avoid flushing at Parquet writes after an exception [parquet-mr]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #1285:
URL: https://github.com/apache/parquet-mr/pull/1285#discussion_r1511370268
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriterError.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.bytes.TrackingByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.filter2.recordlevel.PhoneBookWriter;
+import org.apache.parquet.hadoop.codec.CleanUtil;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.LocalOutputFile;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit test to check how Parquet writing behaves in case of an error happens during the writes. We use an OOM because
+ * that is the most tricky to handle. In this case we shall avoid flushing since it may cause writing to already
+ * released memory spaces.
+ * <p>
+ * To catch the potential issue of writing into released ByteBuffer objects, direct memory allocation is used and at the
+ * release() call we actually release the related direct memory and zero the address inside the ByteBuffer object. As a
+ * result, a subsequent read/write call on the related ByteBuffer object will crash the whole jvm. (Unfortunately, there
+ * is no better way to test this.) To avoid crashing the test executor jvm, the code of this test is executed in a
+ * separate process.
+ */
+public class TestParquetWriterError {
+
+ @Rule
+ public TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ @Test
+ public void testInSeparateProcess() throws IOException, InterruptedException {
+ String outputFile = tmpFolder.newFile("out.parquet").toString();
+
+ String classpath = System.getProperty("java.class.path");
+ String javaPath = Paths.get(System.getProperty("java.home"), "bin", "java")
+ .toAbsolutePath()
+ .toString();
+ Process process = new ProcessBuilder()
+ .command(javaPath, "-cp", classpath, Main.class.getName(), outputFile)
+ .redirectError(ProcessBuilder.Redirect.INHERIT)
+ .redirectOutput(ProcessBuilder.Redirect.INHERIT)
+ .start();
+ Assert.assertEquals(
+ "Test process exited with a non-zero return code. See previous logs for details.",
+ 0,
+ process.waitFor());
+ }
+
+ /**
+ * The class to be used to execute this test in a separate thread.
+ */
+ public static class Main {
+
+ private static final Random RANDOM = new Random(2024_02_27_14_20L);
+
+ // See the release() implementation in createAllocator()
+ private static final Field BUFFER_ADDRESS;
+
+ static {
+ Field bufferAddress;
+ try {
+ Class<?> bufferClass = Class.forName("java.nio.Buffer");
+ bufferAddress = bufferClass.getDeclaredField("address");
+ bufferAddress.setAccessible(true);
+ } catch (Exception e) {
+ // From java 17 it does not work, but we still test on earlier ones, so we are fine
+ bufferAddress = null;
+ }
+ BUFFER_ADDRESS = bufferAddress;
+ }
+
+ private static Group generateNext() {
+ PhoneBookWriter.Location location;
+ double chance = RANDOM.nextDouble();
+ if (chance < .45) {
+ location = new PhoneBookWriter.Location(RANDOM.nextDouble(), RANDOM.nextDouble());
+ } else if (chance < .9) {
+ location = new PhoneBookWriter.Location(RANDOM.nextDouble(), null);
+ } else {
+ location = null;
+ }
+ List<PhoneBookWriter.PhoneNumber> phoneNumbers;
+ if (RANDOM.nextDouble() < .1) {
+ phoneNumbers = null;
+ } else {
+ int n = RANDOM.nextInt(4);
+ phoneNumbers = new ArrayList<>(n);
+ for (int i = 0; i < n; ++i) {
+ String kind = RANDOM.nextDouble() < .1 ? null : "kind" + RANDOM.nextInt(5);
+ phoneNumbers.add(new PhoneBookWriter.PhoneNumber(RANDOM.nextInt(), kind));
+ }
+ }
+ String name = RANDOM.nextDouble() < .1 ? null : "name" + RANDOM.nextLong();
+ PhoneBookWriter.User user = new PhoneBookWriter.User(RANDOM.nextLong(), name, phoneNumbers, location);
+ return PhoneBookWriter.groupFromUser(user);
+ }
+
+ private static TrackingByteBufferAllocator createAllocator(final int oomAt) {
+ return TrackingByteBufferAllocator.wrap(new DirectByteBufferAllocator() {
+ private int counter = 0;
+
+ @Override
+ public ByteBuffer allocate(int size) {
+ if (++counter >= oomAt) {
+ Assert.assertEquals(
+ "There should not be any additional allocations after an OOM", oomAt, counter);
+ throw new OutOfMemoryError("Artificial OOM to fail write");
+ }
+ return super.allocate(size);
+ }
+
+ @Override
+ public void release(ByteBuffer b) {
+ CleanUtil.cleanDirectBuffer(b);
+
+ // It seems, if the size of the buffers are small, the related memory space is not given back to the
+ // OS, so writing to them after release does not cause any identifiable issue. Therefore, we
+ // try to explicitly zero the address, so the jvm crashes for a subsequent access.
+ try {
+ if (BUFFER_ADDRESS != null) {
+ BUFFER_ADDRESS.setLong(b, 0L);
+ }
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException("Unable to zero direct ByteBuffer address", e);
+ }
+ }
+ });
+ }
+
+ public static void main(String[] args) throws Throwable {
+ CompressionCodecName[] codecs = {
+ CompressionCodecName.UNCOMPRESSED,
+ CompressionCodecName.GZIP,
+ CompressionCodecName.SNAPPY,
+ CompressionCodecName.ZSTD,
+ CompressionCodecName.LZ4_RAW
+ };
+ for (int cycle = 0; cycle < 50; ++cycle) {
+ try (TrackingByteBufferAllocator allocator = createAllocator(RANDOM.nextInt(100) + 1);
+ ParquetWriter<Group> writer = ExampleParquetWriter.builder(
+ new LocalOutputFile(Paths.get(args[0])))
+ .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+ .withType(PhoneBookWriter.getSchema())
+ .withAllocator(allocator)
+ .withCodecFactory(CodecFactory.createDirectCodecFactory(
+ new Configuration(), allocator, ParquetProperties.DEFAULT_PAGE_SIZE))
+ .withCompressionCodec(codecs[RANDOM.nextInt(codecs.length)])
Review Comment:
How about adding a comment to explicitly state this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] PARQUET-2437: Avoid flushing at Parquet writes after an exception [parquet-mr]
Posted by "gszadovszky (via GitHub)" <gi...@apache.org>.
gszadovszky commented on code in PR #1285:
URL: https://github.com/apache/parquet-mr/pull/1285#discussion_r1510725271
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriterError.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.bytes.TrackingByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.filter2.recordlevel.PhoneBookWriter;
+import org.apache.parquet.hadoop.codec.CleanUtil;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.LocalOutputFile;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit test to check how Parquet writing behaves in case of an error happens during the writes. We use an OOM because
+ * that is the most tricky to handle. In this case we shall avoid flushing since it may cause writing to already
+ * released memory spaces.
+ * <p>
+ * To catch the potential issue of writing into released ByteBuffer objects, direct memory allocation is used and at the
+ * release() call we actually release the related direct memory and zero the address inside the ByteBuffer object. As a
+ * result, a subsequent read/write call on the related ByteBuffer object will crash the whole jvm. (Unfortunately, there
+ * is no better way to test this.) To avoid crashing the test executor jvm, the code of this test is executed in a
+ * separate process.
+ */
+public class TestParquetWriterError {
+
+ @Rule
+ public TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ @Test
+ public void testInSeparateProcess() throws IOException, InterruptedException {
+ String outputFile = tmpFolder.newFile("out.parquet").toString();
+
+ String classpath = System.getProperty("java.class.path");
+ String javaPath = Paths.get(System.getProperty("java.home"), "bin", "java")
+ .toAbsolutePath()
+ .toString();
+ Process process = new ProcessBuilder()
+ .command(javaPath, "-cp", classpath, Main.class.getName(), outputFile)
+ .redirectError(ProcessBuilder.Redirect.INHERIT)
+ .redirectOutput(ProcessBuilder.Redirect.INHERIT)
+ .start();
+ Assert.assertEquals(
+ "Test process exited with a non-zero return code. See previous logs for details.",
+ 0,
+ process.waitFor());
+ }
+
+ /**
+ * The class to be used to execute this test in a separate thread.
+ */
+ public static class Main {
+
+ private static final Random RANDOM = new Random(2024_02_27_14_20L);
+
+ // See the release() implementation in createAllocator()
+ private static final Field BUFFER_ADDRESS;
+
+ static {
+ Field bufferAddress;
+ try {
+ Class<?> bufferClass = Class.forName("java.nio.Buffer");
+ bufferAddress = bufferClass.getDeclaredField("address");
+ bufferAddress.setAccessible(true);
+ } catch (Exception e) {
+ // From java 17 it does not work, but we still test on earlier ones, so we are fine
Review Comment:
Yes, it fails at `setAccessible(true)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] PARQUET-2437: Avoid flushing at Parquet writes after an exception [parquet-mr]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #1285:
URL: https://github.com/apache/parquet-mr/pull/1285#discussion_r1510254858
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriterError.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.bytes.TrackingByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.filter2.recordlevel.PhoneBookWriter;
+import org.apache.parquet.hadoop.codec.CleanUtil;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.LocalOutputFile;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit test to check how Parquet writing behaves in case of an error happens during the writes. We use an OOM because
+ * that is the most tricky to handle. In this case we shall avoid flushing since it may cause writing to already
+ * released memory spaces.
+ * <p>
+ * To catch the potential issue of writing into released ByteBuffer objects, direct memory allocation is used and at the
+ * release() call we actually release the related direct memory and zero the address inside the ByteBuffer object. As a
+ * result, a subsequent read/write call on the related ByteBuffer object will crash the whole jvm. (Unfortunately, there
+ * is no better way to test this.) To avoid crashing the test executor jvm, the code of this test is executed in a
+ * separate process.
+ */
+public class TestParquetWriterError {
+
+ @Rule
+ public TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ @Test
+ public void testInSeparateProcess() throws IOException, InterruptedException {
+ String outputFile = tmpFolder.newFile("out.parquet").toString();
+
+ String classpath = System.getProperty("java.class.path");
+ String javaPath = Paths.get(System.getProperty("java.home"), "bin", "java")
+ .toAbsolutePath()
+ .toString();
+ Process process = new ProcessBuilder()
+ .command(javaPath, "-cp", classpath, Main.class.getName(), outputFile)
+ .redirectError(ProcessBuilder.Redirect.INHERIT)
+ .redirectOutput(ProcessBuilder.Redirect.INHERIT)
+ .start();
+ Assert.assertEquals(
+ "Test process exited with a non-zero return code. See previous logs for details.",
+ 0,
+ process.waitFor());
+ }
+
+ /**
+ * The class to be used to execute this test in a separate thread.
+ */
+ public static class Main {
+
+ private static final Random RANDOM = new Random(2024_02_27_14_20L);
+
+ // See the release() implementation in createAllocator()
+ private static final Field BUFFER_ADDRESS;
+
+ static {
+ Field bufferAddress;
+ try {
+ Class<?> bufferClass = Class.forName("java.nio.Buffer");
+ bufferAddress = bufferClass.getDeclaredField("address");
+ bufferAddress.setAccessible(true);
+ } catch (Exception e) {
+ // From java 17 it does not work, but we still test on earlier ones, so we are fine
Review Comment:
Do you mean this test will fall on java 17?
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriterError.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.bytes.TrackingByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.filter2.recordlevel.PhoneBookWriter;
+import org.apache.parquet.hadoop.codec.CleanUtil;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.LocalOutputFile;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit test to check how Parquet writing behaves in case of an error happens during the writes. We use an OOM because
+ * that is the most tricky to handle. In this case we shall avoid flushing since it may cause writing to already
+ * released memory spaces.
+ * <p>
+ * To catch the potential issue of writing into released ByteBuffer objects, direct memory allocation is used and at the
+ * release() call we actually release the related direct memory and zero the address inside the ByteBuffer object. As a
+ * result, a subsequent read/write call on the related ByteBuffer object will crash the whole jvm. (Unfortunately, there
+ * is no better way to test this.) To avoid crashing the test executor jvm, the code of this test is executed in a
+ * separate process.
+ */
+public class TestParquetWriterError {
+
+ @Rule
+ public TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ @Test
+ public void testInSeparateProcess() throws IOException, InterruptedException {
+ String outputFile = tmpFolder.newFile("out.parquet").toString();
+
+ String classpath = System.getProperty("java.class.path");
+ String javaPath = Paths.get(System.getProperty("java.home"), "bin", "java")
+ .toAbsolutePath()
+ .toString();
+ Process process = new ProcessBuilder()
+ .command(javaPath, "-cp", classpath, Main.class.getName(), outputFile)
+ .redirectError(ProcessBuilder.Redirect.INHERIT)
+ .redirectOutput(ProcessBuilder.Redirect.INHERIT)
+ .start();
+ Assert.assertEquals(
+ "Test process exited with a non-zero return code. See previous logs for details.",
+ 0,
+ process.waitFor());
+ }
+
+ /**
+ * The class to be used to execute this test in a separate thread.
+ */
+ public static class Main {
+
+ private static final Random RANDOM = new Random(2024_02_27_14_20L);
Review Comment:
I'm not sure if it is easy to figure out a set of test data to get rid of the random approach, thus we may reduce the run time and possibly flaky failure.
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriterError.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.bytes.TrackingByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.filter2.recordlevel.PhoneBookWriter;
+import org.apache.parquet.hadoop.codec.CleanUtil;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.LocalOutputFile;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit test to check how Parquet writing behaves in case of an error happens during the writes. We use an OOM because
+ * that is the most tricky to handle. In this case we shall avoid flushing since it may cause writing to already
+ * released memory spaces.
+ * <p>
+ * To catch the potential issue of writing into released ByteBuffer objects, direct memory allocation is used and at the
+ * release() call we actually release the related direct memory and zero the address inside the ByteBuffer object. As a
+ * result, a subsequent read/write call on the related ByteBuffer object will crash the whole jvm. (Unfortunately, there
+ * is no better way to test this.) To avoid crashing the test executor jvm, the code of this test is executed in a
+ * separate process.
+ */
+public class TestParquetWriterError {
+
+ @Rule
+ public TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ @Test
+ public void testInSeparateProcess() throws IOException, InterruptedException {
+ String outputFile = tmpFolder.newFile("out.parquet").toString();
+
+ String classpath = System.getProperty("java.class.path");
+ String javaPath = Paths.get(System.getProperty("java.home"), "bin", "java")
+ .toAbsolutePath()
+ .toString();
+ Process process = new ProcessBuilder()
+ .command(javaPath, "-cp", classpath, Main.class.getName(), outputFile)
+ .redirectError(ProcessBuilder.Redirect.INHERIT)
+ .redirectOutput(ProcessBuilder.Redirect.INHERIT)
+ .start();
+ Assert.assertEquals(
+ "Test process exited with a non-zero return code. See previous logs for details.",
+ 0,
+ process.waitFor());
+ }
+
+ /**
+ * The class to be used to execute this test in a separate thread.
+ */
+ public static class Main {
+
+ private static final Random RANDOM = new Random(2024_02_27_14_20L);
+
+ // See the release() implementation in createAllocator()
+ private static final Field BUFFER_ADDRESS;
+
+ static {
+ Field bufferAddress;
+ try {
+ Class<?> bufferClass = Class.forName("java.nio.Buffer");
+ bufferAddress = bufferClass.getDeclaredField("address");
+ bufferAddress.setAccessible(true);
+ } catch (Exception e) {
+ // From java 17 it does not work, but we still test on earlier ones, so we are fine
+ bufferAddress = null;
+ }
+ BUFFER_ADDRESS = bufferAddress;
+ }
+
+ private static Group generateNext() {
+ PhoneBookWriter.Location location;
+ double chance = RANDOM.nextDouble();
+ if (chance < .45) {
+ location = new PhoneBookWriter.Location(RANDOM.nextDouble(), RANDOM.nextDouble());
+ } else if (chance < .9) {
+ location = new PhoneBookWriter.Location(RANDOM.nextDouble(), null);
+ } else {
+ location = null;
+ }
+ List<PhoneBookWriter.PhoneNumber> phoneNumbers;
+ if (RANDOM.nextDouble() < .1) {
+ phoneNumbers = null;
+ } else {
+ int n = RANDOM.nextInt(4);
+ phoneNumbers = new ArrayList<>(n);
+ for (int i = 0; i < n; ++i) {
+ String kind = RANDOM.nextDouble() < .1 ? null : "kind" + RANDOM.nextInt(5);
+ phoneNumbers.add(new PhoneBookWriter.PhoneNumber(RANDOM.nextInt(), kind));
+ }
+ }
+ String name = RANDOM.nextDouble() < .1 ? null : "name" + RANDOM.nextLong();
+ PhoneBookWriter.User user = new PhoneBookWriter.User(RANDOM.nextLong(), name, phoneNumbers, location);
+ return PhoneBookWriter.groupFromUser(user);
+ }
+
+ private static TrackingByteBufferAllocator createAllocator(final int oomAt) {
+ return TrackingByteBufferAllocator.wrap(new DirectByteBufferAllocator() {
+ private int counter = 0;
+
+ @Override
+ public ByteBuffer allocate(int size) {
+ if (++counter >= oomAt) {
+ Assert.assertEquals(
+ "There should not be any additional allocations after an OOM", oomAt, counter);
+ throw new OutOfMemoryError("Artificial OOM to fail write");
+ }
+ return super.allocate(size);
+ }
+
+ @Override
+ public void release(ByteBuffer b) {
+ CleanUtil.cleanDirectBuffer(b);
+
+ // It seems, if the size of the buffers are small, the related memory space is not given back to the
+ // OS, so writing to them after release does not cause any identifiable issue. Therefore, we
+ // try to explicitly zero the address, so the jvm crashes for a subsequent access.
+ try {
+ if (BUFFER_ADDRESS != null) {
+ BUFFER_ADDRESS.setLong(b, 0L);
+ }
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException("Unable to zero direct ByteBuffer address", e);
+ }
+ }
+ });
+ }
+
+ public static void main(String[] args) throws Throwable {
+ CompressionCodecName[] codecs = {
+ CompressionCodecName.UNCOMPRESSED,
+ CompressionCodecName.GZIP,
+ CompressionCodecName.SNAPPY,
+ CompressionCodecName.ZSTD,
+ CompressionCodecName.LZ4_RAW
+ };
+ for (int cycle = 0; cycle < 50; ++cycle) {
+ try (TrackingByteBufferAllocator allocator = createAllocator(RANDOM.nextInt(100) + 1);
+ ParquetWriter<Group> writer = ExampleParquetWriter.builder(
+ new LocalOutputFile(Paths.get(args[0])))
+ .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+ .withType(PhoneBookWriter.getSchema())
+ .withAllocator(allocator)
+ .withCodecFactory(CodecFactory.createDirectCodecFactory(
+ new Configuration(), allocator, ParquetProperties.DEFAULT_PAGE_SIZE))
+ .withCompressionCodec(codecs[RANDOM.nextInt(codecs.length)])
Review Comment:
Why do we need to change the codec?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] PARQUET-2437: Avoid flushing at Parquet writes after an exception [parquet-mr]
Posted by "gszadovszky (via GitHub)" <gi...@apache.org>.
gszadovszky commented on code in PR #1285:
URL: https://github.com/apache/parquet-mr/pull/1285#discussion_r1510727631
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriterError.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.bytes.TrackingByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.filter2.recordlevel.PhoneBookWriter;
+import org.apache.parquet.hadoop.codec.CleanUtil;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.LocalOutputFile;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit test to check how Parquet writing behaves in case of an error happens during the writes. We use an OOM because
+ * that is the most tricky to handle. In this case we shall avoid flushing since it may cause writing to already
+ * released memory spaces.
+ * <p>
+ * To catch the potential issue of writing into released ByteBuffer objects, direct memory allocation is used and at the
+ * release() call we actually release the related direct memory and zero the address inside the ByteBuffer object. As a
+ * result, a subsequent read/write call on the related ByteBuffer object will crash the whole jvm. (Unfortunately, there
+ * is no better way to test this.) To avoid crashing the test executor jvm, the code of this test is executed in a
+ * separate process.
+ */
+public class TestParquetWriterError {
+
+ @Rule
+ public TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ @Test
+ public void testInSeparateProcess() throws IOException, InterruptedException {
+ String outputFile = tmpFolder.newFile("out.parquet").toString();
+
+ String classpath = System.getProperty("java.class.path");
+ String javaPath = Paths.get(System.getProperty("java.home"), "bin", "java")
+ .toAbsolutePath()
+ .toString();
+ Process process = new ProcessBuilder()
+ .command(javaPath, "-cp", classpath, Main.class.getName(), outputFile)
+ .redirectError(ProcessBuilder.Redirect.INHERIT)
+ .redirectOutput(ProcessBuilder.Redirect.INHERIT)
+ .start();
+ Assert.assertEquals(
+ "Test process exited with a non-zero return code. See previous logs for details.",
+ 0,
+ process.waitFor());
+ }
+
+ /**
+ * The class to be used to execute this test in a separate thread.
+ */
+ public static class Main {
+
+ private static final Random RANDOM = new Random(2024_02_27_14_20L);
+
+ // See the release() implementation in createAllocator()
+ private static final Field BUFFER_ADDRESS;
+
+ static {
+ Field bufferAddress;
+ try {
+ Class<?> bufferClass = Class.forName("java.nio.Buffer");
+ bufferAddress = bufferClass.getDeclaredField("address");
+ bufferAddress.setAccessible(true);
+ } catch (Exception e) {
+ // From java 17 it does not work, but we still test on earlier ones, so we are fine
+ bufferAddress = null;
+ }
+ BUFFER_ADDRESS = bufferAddress;
+ }
+
+ private static Group generateNext() {
+ PhoneBookWriter.Location location;
+ double chance = RANDOM.nextDouble();
+ if (chance < .45) {
+ location = new PhoneBookWriter.Location(RANDOM.nextDouble(), RANDOM.nextDouble());
+ } else if (chance < .9) {
+ location = new PhoneBookWriter.Location(RANDOM.nextDouble(), null);
+ } else {
+ location = null;
+ }
+ List<PhoneBookWriter.PhoneNumber> phoneNumbers;
+ if (RANDOM.nextDouble() < .1) {
+ phoneNumbers = null;
+ } else {
+ int n = RANDOM.nextInt(4);
+ phoneNumbers = new ArrayList<>(n);
+ for (int i = 0; i < n; ++i) {
+ String kind = RANDOM.nextDouble() < .1 ? null : "kind" + RANDOM.nextInt(5);
+ phoneNumbers.add(new PhoneBookWriter.PhoneNumber(RANDOM.nextInt(), kind));
+ }
+ }
+ String name = RANDOM.nextDouble() < .1 ? null : "name" + RANDOM.nextLong();
+ PhoneBookWriter.User user = new PhoneBookWriter.User(RANDOM.nextLong(), name, phoneNumbers, location);
+ return PhoneBookWriter.groupFromUser(user);
+ }
+
+ private static TrackingByteBufferAllocator createAllocator(final int oomAt) {
+ return TrackingByteBufferAllocator.wrap(new DirectByteBufferAllocator() {
+ private int counter = 0;
+
+ @Override
+ public ByteBuffer allocate(int size) {
+ if (++counter >= oomAt) {
+ Assert.assertEquals(
+ "There should not be any additional allocations after an OOM", oomAt, counter);
+ throw new OutOfMemoryError("Artificial OOM to fail write");
+ }
+ return super.allocate(size);
+ }
+
+ @Override
+ public void release(ByteBuffer b) {
+ CleanUtil.cleanDirectBuffer(b);
+
+ // It seems, if the size of the buffers are small, the related memory space is not given back to the
+ // OS, so writing to them after release does not cause any identifiable issue. Therefore, we
+ // try to explicitly zero the address, so the jvm crashes for a subsequent access.
+ try {
+ if (BUFFER_ADDRESS != null) {
+ BUFFER_ADDRESS.setLong(b, 0L);
+ }
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException("Unable to zero direct ByteBuffer address", e);
+ }
+ }
+ });
+ }
+
+ public static void main(String[] args) throws Throwable {
+ CompressionCodecName[] codecs = {
+ CompressionCodecName.UNCOMPRESSED,
+ CompressionCodecName.GZIP,
+ CompressionCodecName.SNAPPY,
+ CompressionCodecName.ZSTD,
+ CompressionCodecName.LZ4_RAW
+ };
+ for (int cycle = 0; cycle < 50; ++cycle) {
+ try (TrackingByteBufferAllocator allocator = createAllocator(RANDOM.nextInt(100) + 1);
+ ParquetWriter<Group> writer = ExampleParquetWriter.builder(
+ new LocalOutputFile(Paths.get(args[0])))
+ .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+ .withType(PhoneBookWriter.getSchema())
+ .withAllocator(allocator)
+ .withCodecFactory(CodecFactory.createDirectCodecFactory(
+ new Configuration(), allocator, ParquetProperties.DEFAULT_PAGE_SIZE))
+ .withCompressionCodec(codecs[RANDOM.nextInt(codecs.length)])
Review Comment:
I am using direct codec factory here with different codecs which also have direct memory allocated. I wanted to be sure that those are not failing in this scenario either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] PARQUET-2437: Avoid flushing at Parquet writes after an exception [parquet-mr]
Posted by "gszadovszky (via GitHub)" <gi...@apache.org>.
gszadovszky commented on code in PR #1285:
URL: https://github.com/apache/parquet-mr/pull/1285#discussion_r1510731818
##########
parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriterError.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.bytes.TrackingByteBufferAllocator;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.filter2.recordlevel.PhoneBookWriter;
+import org.apache.parquet.hadoop.codec.CleanUtil;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.io.LocalOutputFile;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit test to check how Parquet writing behaves in case of an error happens during the writes. We use an OOM because
+ * that is the most tricky to handle. In this case we shall avoid flushing since it may cause writing to already
+ * released memory spaces.
+ * <p>
+ * To catch the potential issue of writing into released ByteBuffer objects, direct memory allocation is used and at the
+ * release() call we actually release the related direct memory and zero the address inside the ByteBuffer object. As a
+ * result, a subsequent read/write call on the related ByteBuffer object will crash the whole jvm. (Unfortunately, there
+ * is no better way to test this.) To avoid crashing the test executor jvm, the code of this test is executed in a
+ * separate process.
+ */
+public class TestParquetWriterError {
+
+ @Rule
+ public TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ @Test
+ public void testInSeparateProcess() throws IOException, InterruptedException {
+ String outputFile = tmpFolder.newFile("out.parquet").toString();
+
+ String classpath = System.getProperty("java.class.path");
+ String javaPath = Paths.get(System.getProperty("java.home"), "bin", "java")
+ .toAbsolutePath()
+ .toString();
+ Process process = new ProcessBuilder()
+ .command(javaPath, "-cp", classpath, Main.class.getName(), outputFile)
+ .redirectError(ProcessBuilder.Redirect.INHERIT)
+ .redirectOutput(ProcessBuilder.Redirect.INHERIT)
+ .start();
+ Assert.assertEquals(
+ "Test process exited with a non-zero return code. See previous logs for details.",
+ 0,
+ process.waitFor());
+ }
+
+ /**
+ * The class to be used to execute this test in a separate thread.
+ */
+ public static class Main {
+
+ private static final Random RANDOM = new Random(2024_02_27_14_20L);
Review Comment:
This is pseudo random with a fixed seed. It means we use the very same data and scenario. 100% reproducible. I am using random because it is simpler and shorter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] PARQUET-2437: Avoid flushing at Parquet writes after an exception [parquet-mr]
Posted by "gszadovszky (via GitHub)" <gi...@apache.org>.
gszadovszky merged PR #1285:
URL: https://github.com/apache/parquet-mr/pull/1285
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org