You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@parquet.apache.org by GitBox <gi...@apache.org> on 2021/04/19 12:11:54 UTC
[GitHub] [parquet-mr] Miksu82 opened a new pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Miksu82 opened a new pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895
This MR fixes Parquet-2030. It exposes `page.size.row.check.max` and `page.size.row.check.max` to `ParquetWriter.Builder`
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] Miksu82 commented on a change in pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
Miksu82 commented on a change in pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#discussion_r616693569
##########
File path: parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriter.java
##########
@@ -279,4 +285,133 @@ public void testParquetFileWithBloomFilter() throws IOException {
LongHashFunction.xx(0).hashBytes(Binary.fromString(name).toByteBuffer())));
}
}
+
+ @Test
+ public void testParquetFileNotFlushedWhenRowCountLimitsAreNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedWhenMinRowCountIsExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(3)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 3, 1);
+ }
+
+ @Test
+ public void testParquetFileIsNotFlushedIfMinRowCountIsNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(2)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedAfterEachRecord() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(1)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 3, 3);
+ }
+
+ @Test
+ public void testParquetFileNotFlushingAllRows() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(2)
+ .withMaxRowCountForPageSizeCheck(3)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 2, 1);
+ }
+
+ private ExampleParquetWriter.Builder getParquetWriterBuilder(MessageType schema,
+ TestOutputFile file) throws IOException {
+ Configuration conf = new Configuration();
+ GroupWriteSupport.setSchema(schema, conf);
+ return ExampleParquetWriter.builder(file)
+ .withConf(conf)
+ // Set row group size to 1, to make sure we flush every time
+ // minRowCountForPageSizeCheck or maxRowCountForPageSizeCheck is exceeded
+ .withRowGroupSize(1);
+ }
+
+ private void writeRecords(ParquetWriter<Group> writer, MessageType schema) throws IOException {
+ SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+ writer.write(factory.newGroup().append("str", "foo"));
+ writer.write(factory.newGroup().append("str", "bar"));
+ writer.write(factory.newGroup().append("str", "baz"));
+ }
+
+ private void assertHasNotFlushed(TestOutputFile file) {
+ int emptyFileLength = ParquetFileWriter.MAGIC.length;
+ assertEquals(emptyFileLength, file.stats.getBytesWritten());
+ }
+
+ private void assertHasFlushed(TestOutputFile file, int numWrites, int numFlushes) {
Review comment:
Seems really hard to assert that the properties are just set. See my comment below
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] Miksu82 edited a comment on pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
Miksu82 edited a comment on pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#issuecomment-823282453
Unfortunately I cannot find a way to only assert that properties have been set. It probably could be done with extensive refactoring but I don't see the point to do that in this PR. Also other property setters in the builder are not unit tested so I guess I could just remove these tests completely.
However I did notice that I can assert the number of flushes by counting the number of blocks in the Parquet file footer. Asserting that way would make the test easier to understand.
Let me know what you prefer. Remove the tests or fix the assertations? I'll continue after that.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] gszadovszky commented on a change in pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
gszadovszky commented on a change in pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#discussion_r616573508
##########
File path: parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriter.java
##########
@@ -279,4 +285,133 @@ public void testParquetFileWithBloomFilter() throws IOException {
LongHashFunction.xx(0).hashBytes(Binary.fromString(name).toByteBuffer())));
}
}
+
+ @Test
+ public void testParquetFileNotFlushedWhenRowCountLimitsAreNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedWhenMinRowCountIsExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(3)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 3, 1);
+ }
+
+ @Test
+ public void testParquetFileIsNotFlushedIfMinRowCountIsNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(2)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedAfterEachRecord() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(1)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 3, 3);
+ }
+
+ @Test
+ public void testParquetFileNotFlushingAllRows() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(2)
+ .withMaxRowCountForPageSizeCheck(3)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 2, 1);
+ }
+
+ private ExampleParquetWriter.Builder getParquetWriterBuilder(MessageType schema,
+ TestOutputFile file) throws IOException {
+ Configuration conf = new Configuration();
+ GroupWriteSupport.setSchema(schema, conf);
+ return ExampleParquetWriter.builder(file)
+ .withConf(conf)
+ // Set row group size to 1, to make sure we flush every time
+ // minRowCountForPageSizeCheck or maxRowCountForPageSizeCheck is exceeded
+ .withRowGroupSize(1);
+ }
+
+ private void writeRecords(ParquetWriter<Group> writer, MessageType schema) throws IOException {
+ SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+ writer.write(factory.newGroup().append("str", "foo"));
+ writer.write(factory.newGroup().append("str", "bar"));
+ writer.write(factory.newGroup().append("str", "baz"));
+ }
+
+ private void assertHasNotFlushed(TestOutputFile file) {
+ int emptyFileLength = ParquetFileWriter.MAGIC.length;
+ assertEquals(emptyFileLength, file.stats.getBytesWritten());
+ }
+
+ private void assertHasFlushed(TestOutputFile file, int numWrites, int numFlushes) {
Review comment:
The method name suggests that it validates if the file was flushed `numFlushes` times. But it is executed only once after all writes so it only validates that it flushed once and all data is written.
##########
File path: parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriter.java
##########
@@ -279,4 +285,133 @@ public void testParquetFileWithBloomFilter() throws IOException {
LongHashFunction.xx(0).hashBytes(Binary.fromString(name).toByteBuffer())));
}
}
+
+ @Test
+ public void testParquetFileNotFlushedWhenRowCountLimitsAreNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedWhenMinRowCountIsExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(3)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 3, 1);
+ }
+
+ @Test
+ public void testParquetFileIsNotFlushedIfMinRowCountIsNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(2)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedAfterEachRecord() throws IOException {
Review comment:
This test does not validate each flush but only one at the end. See my last comment.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] Miksu82 commented on pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
Miksu82 commented on pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#issuecomment-823282453
Unfortunately I cannot find a way to only assert that properties have been set. It probably could be done with extensive refactoring but I don't see the point to do that in this test. Also other property setters in the builder are not unit tested so I guess I could just remove these tests completely.
However I did notice that I can assert the number of flushes by counting the number of blocks in the Parquet file footer. Asserting that way would make the test easier to understand.
Let me know what you prefer. Remove the tests or fix the assertations? I'll continue after that.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] gszadovszky commented on pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
gszadovszky commented on pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#issuecomment-823315022
Yeah, checking only that properties are really passed is not easy. I did not want to say to remove all the tests completely (or any) just that we don't need to test every aspect of the original feature. I think, it is enough to test the difference between setting the properties and not so you know they are passed. You can see that passing such properties are not that obvious so I would keep some test to be sure.
The idea of checking the number of row groups is a good point.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] gszadovszky merged pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
gszadovszky merged pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] Miksu82 commented on pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
Miksu82 commented on pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#issuecomment-823879944
I now have two tests. One with default params and one with changes params. And I assert by checking the number of blocks in the footer.
I also removed the changes I did for `TestOutputFile` since they are not needed anymore.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] Miksu82 commented on a change in pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
Miksu82 commented on a change in pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#discussion_r616653104
##########
File path: parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriter.java
##########
@@ -279,4 +285,133 @@ public void testParquetFileWithBloomFilter() throws IOException {
LongHashFunction.xx(0).hashBytes(Binary.fromString(name).toByteBuffer())));
}
}
+
+ @Test
+ public void testParquetFileNotFlushedWhenRowCountLimitsAreNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedWhenMinRowCountIsExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(3)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 3, 1);
+ }
+
+ @Test
+ public void testParquetFileIsNotFlushedIfMinRowCountIsNotExceeded() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(4)
+ .withMaxRowCountForPageSizeCheck(2)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasNotFlushed(file);
+ }
+
+ @Test
+ public void testParquetFileIsFlushedAfterEachRecord() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(1)
+ .withMaxRowCountForPageSizeCheck(4)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 3, 3);
+ }
+
+ @Test
+ public void testParquetFileNotFlushingAllRows() throws IOException {
+ MessageType schema = Types
+ .buildMessage()
+ .required(BINARY)
+ .as(stringType())
+ .named("str")
+ .named("msg");
+
+ TestOutputFile file = new TestOutputFile();
+ ParquetWriter<Group> writer = getParquetWriterBuilder(schema, file)
+ .withMinRowCountForPageSizeCheck(2)
+ .withMaxRowCountForPageSizeCheck(3)
+ .build();
+
+ writeRecords(writer, schema);
+
+ assertHasFlushed(file, 2, 1);
+ }
+
+ private ExampleParquetWriter.Builder getParquetWriterBuilder(MessageType schema,
+ TestOutputFile file) throws IOException {
+ Configuration conf = new Configuration();
+ GroupWriteSupport.setSchema(schema, conf);
+ return ExampleParquetWriter.builder(file)
+ .withConf(conf)
+ // Set row group size to 1, to make sure we flush every time
+ // minRowCountForPageSizeCheck or maxRowCountForPageSizeCheck is exceeded
+ .withRowGroupSize(1);
+ }
+
+ private void writeRecords(ParquetWriter<Group> writer, MessageType schema) throws IOException {
+ SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+ writer.write(factory.newGroup().append("str", "foo"));
+ writer.write(factory.newGroup().append("str", "bar"));
+ writer.write(factory.newGroup().append("str", "baz"));
+ }
+
+ private void assertHasNotFlushed(TestOutputFile file) {
+ int emptyFileLength = ParquetFileWriter.MAGIC.length;
+ assertEquals(emptyFileLength, file.stats.getBytesWritten());
+ }
+
+ private void assertHasFlushed(TestOutputFile file, int numWrites, int numFlushes) {
Review comment:
Hmm..no. Everytime a record is flushed it writes the actual record + the metadata to indicate that a row group has ended (the writer is created by defining `rowGroupSize=1`).
It works like this
```
write("foo") // writes 7B
flush() // writes 23B
write("bar") // writes 7B
flush() // writes 23B
// Totally has written: 60B
```
but
```
write("foo") // writes 7B
write("bar") // writes 7B
flush() // writes 23B
// Totally has written: 37
```
And by defining the `page.size.row.check.min` and `page.size.row.check.max` I can control how often the `InternalParquetRecordWriter` flushes the records, and I can assert that by calculating the number of bytes written in total.
But you are absolutely right, that this is complicated and in the wrong place so I'll remove these test and just assert that the properties have been set.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [parquet-mr] Miksu82 commented on pull request #895: Parquet-2030: Expose page size row check configurations to ParquetWri…
Posted by GitBox <gi...@apache.org>.
Miksu82 commented on pull request #895:
URL: https://github.com/apache/parquet-mr/pull/895#issuecomment-823242208
> We do not need to validate the whole feature of the page size check only the fact that the properties have passed properly
Yeah you are right, that should be in `InternalParquetRecordWriter` unit tests. I'll try to simplify the tests I've added so that they only test that the property has been set.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org