You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/10/14 13:14:05 UTC
[GitHub] [iceberg] NeQuissimus opened a new pull request #3293: Lazily initialize the writer inside ParquetWriter
NeQuissimus opened a new pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293
Noticed that the inner writer is not needed during construction and in add().
Allocating it can - depending on the selected implementation - cause many other
objects to be initialized.
This is particularly true with the Hadoop GCS connector, which immediately
allocates a 64 MiB `byte[]`.
As long as code is tight-looping over the `add()` method while adding additional
data rows to the parquet file, there is no reason why
`ParquetWriter` needs to initialize these instances.
All usages of `this.writer` have been changed to use `getWriter()` instead.
`getWriter()` will allocate a single writer object when needed.
Also improved `startRowGroup()`, which previously needed access to the writer
object but no longer does.
The only call was to `nextRowGroupSize`, which returns the result from an
`AlignmentStrategy`.
However, Iceberg's code initializes a `ParquetFileWriter` with a `maxPaddingSize
= 0`. No matter the particular alignment strategy, this will cause
`rowGroupSize` to always be returns as the value of `nextRowGroupSize`.
Hence, we can remove the indirection through the writer object entirely.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r732036958
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -183,6 +184,16 @@ public WriteBuilder writerVersion(WriterVersion version) {
return this;
}
+ public WriteBuilder lazy() {
+ this.lazyWriter = true;
+ return this;
+ }
+
+ public WriteBuilder lazy(boolean lazy) {
+ this.lazyWriter = lazy;
+ return this;
+ }
+
Review comment:
I think we should remove this. Let's not change the API until someone needs a non-lazy writer.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-947971006
I think the test is still valid. An aborted commit should drop the data files. The way to fix it is probably to write some content to the Parquet file before aborting, if possible. That way it still tests what you want it to.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r735037707
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
The partitioning logic seems to require 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-947959498
@rdblue right, that's what I figured. Does that not mean the entire test is pointless now? The folder/files will not exist before or after calling `abort` since the (now) lazy writer never even created them.
I can change the test to ensure that but it won't really check any of `abort`'s behaviour, will it?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r729972467
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -187,11 +215,6 @@ private void flushRowGroup(boolean finished) {
private void startRowGroup() {
Preconditions.checkState(!closed, "Writer is closed");
- try {
- this.nextRowGroupSize = Math.min(writer.getNextRowGroupSize(), targetRowGroupSize);
- } catch (IOException e) {
- throw new RuntimeIOException(e);
- }
Review comment:
This looks suspicious. Why did you remove it?
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -96,21 +99,46 @@
this.model = (ParquetValueWriter<T>) createWriterFunc.apply(parquetSchema);
this.metricsConfig = metricsConfig;
this.columnIndexTruncateLength = conf.getInt(COLUMN_INDEX_TRUNCATE_LENGTH, DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH);
+ this.writeMode = writeMode;
+ this.output = output;
+ this.conf = conf;
- try {
- this.writer = new ParquetFileWriter(ParquetIO.file(output, conf), parquetSchema,
- writeMode, rowGroupSize, 0);
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to create Parquet file");
+ startRowGroup();
+
+ if (!lazy) {
+ this.getWriter();
}
+ }
- try {
- writer.start();
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to start Parquet file writer");
+ @SuppressWarnings("unchecked")
+ ParquetWriter(Configuration conf, OutputFile output, Schema schema, long rowGroupSize,
+ Map<String, String> metadata,
+ Function<MessageType, ParquetValueWriter<?>> createWriterFunc,
+ CompressionCodecName codec,
+ ParquetProperties properties,
+ MetricsConfig metricsConfig,
+ ParquetFileWriter.Mode writeMode) {
+ this(conf, output, schema, rowGroupSize, metadata, createWriterFunc,
+ codec, properties, metricsConfig, writeMode, false);
+ }
+
+ private ParquetFileWriter getWriter() {
Review comment:
No need for `get` since this returns a writer. We avoid `get` because it doesn't carry any useful information. In this case you can omit it but in most you would want to use a more descriptive verb.
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -96,21 +99,46 @@
this.model = (ParquetValueWriter<T>) createWriterFunc.apply(parquetSchema);
this.metricsConfig = metricsConfig;
this.columnIndexTruncateLength = conf.getInt(COLUMN_INDEX_TRUNCATE_LENGTH, DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH);
+ this.writeMode = writeMode;
+ this.output = output;
+ this.conf = conf;
- try {
- this.writer = new ParquetFileWriter(ParquetIO.file(output, conf), parquetSchema,
- writeMode, rowGroupSize, 0);
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to create Parquet file");
+ startRowGroup();
+
+ if (!lazy) {
+ this.getWriter();
}
+ }
- try {
- writer.start();
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to start Parquet file writer");
+ @SuppressWarnings("unchecked")
+ ParquetWriter(Configuration conf, OutputFile output, Schema schema, long rowGroupSize,
+ Map<String, String> metadata,
+ Function<MessageType, ParquetValueWriter<?>> createWriterFunc,
+ CompressionCodecName codec,
+ ParquetProperties properties,
+ MetricsConfig metricsConfig,
+ ParquetFileWriter.Mode writeMode) {
+ this(conf, output, schema, rowGroupSize, metadata, createWriterFunc,
+ codec, properties, metricsConfig, writeMode, false);
Review comment:
Nit: continuing indentation is 2 indents / 4 spaces, not 8.
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -96,21 +99,46 @@
this.model = (ParquetValueWriter<T>) createWriterFunc.apply(parquetSchema);
this.metricsConfig = metricsConfig;
this.columnIndexTruncateLength = conf.getInt(COLUMN_INDEX_TRUNCATE_LENGTH, DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH);
+ this.writeMode = writeMode;
+ this.output = output;
+ this.conf = conf;
- try {
- this.writer = new ParquetFileWriter(ParquetIO.file(output, conf), parquetSchema,
- writeMode, rowGroupSize, 0);
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to create Parquet file");
+ startRowGroup();
+
+ if (!lazy) {
+ this.getWriter();
}
+ }
- try {
- writer.start();
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to start Parquet file writer");
+ @SuppressWarnings("unchecked")
+ ParquetWriter(Configuration conf, OutputFile output, Schema schema, long rowGroupSize,
+ Map<String, String> metadata,
+ Function<MessageType, ParquetValueWriter<?>> createWriterFunc,
+ CompressionCodecName codec,
+ ParquetProperties properties,
+ MetricsConfig metricsConfig,
+ ParquetFileWriter.Mode writeMode) {
+ this(conf, output, schema, rowGroupSize, metadata, createWriterFunc,
+ codec, properties, metricsConfig, writeMode, false);
+ }
+
+ private ParquetFileWriter getWriter() {
+ if (this.writer == null) {
+ try {
+ this.writer = new ParquetFileWriter(ParquetIO.file(output, conf), parquetSchema,
+ writeMode, targetRowGroupSize, 0);
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to create Parquet file");
+ }
+
+ try {
+ writer.start();
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to start Parquet file writer");
+ }
}
- startRowGroup();
+ return this.writer;
Review comment:
No need to use `this.` when getting the value of a field, only when setting a field.
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -183,6 +184,16 @@ public WriteBuilder writerVersion(WriterVersion version) {
return this;
}
+ public WriteBuilder lazy() {
+ this.lazyWriter = true;
+ return this;
+ }
+
+ public WriteBuilder lazy(boolean lazy) {
+ this.lazyWriter = lazy;
+ return this;
+ }
+
Review comment:
I don't think that it is worth exposing this as an option. We should either lazily create the writer or always create it.
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -187,11 +215,6 @@ private void flushRowGroup(boolean finished) {
private void startRowGroup() {
Preconditions.checkState(!closed, "Writer is closed");
- try {
- this.nextRowGroupSize = Math.min(writer.getNextRowGroupSize(), targetRowGroupSize);
- } catch (IOException e) {
- throw new RuntimeIOException(e);
- }
Review comment:
Thanks, sorry that I missed that originally.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus edited a comment on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus edited a comment on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-946973951
@rdblue Could you take a look at the test failures with me?
Both Flink tests that run `abort` use this code: https://github.com/apache/iceberg/blob/3763952c7b5d101e2c84efb1651b88439f641099/flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java#L226-L229
With the writer here being lazy, it will not create the files the tests check for. And the actual writer is hidden away. I am unsure of the consequences, just a test with an invalid assumption?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-947972674
@rdblue Makes sense, I'll try to get to it tomorrow
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r742130858
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
Sorry, I'm not sure I understand what's happening. Does the test fail? What is the complete error message?
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
Sorry, I'm not sure I understand what's happening. Does the test fail? What is the complete error message?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r729981179
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -187,11 +215,6 @@ private void flushRowGroup(boolean finished) {
private void startRowGroup() {
Preconditions.checkState(!closed, "Writer is closed");
- try {
- this.nextRowGroupSize = Math.min(writer.getNextRowGroupSize(), targetRowGroupSize);
- } catch (IOException e) {
- throw new RuntimeIOException(e);
- }
Review comment:
I tried to explain this in the description. The way the alignment works inside iceberg, this is not needed because nextRowGroupSize will always be identical to targetRowGroupSize due to the maxPadding setting of 0.
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -183,6 +184,16 @@ public WriteBuilder writerVersion(WriterVersion version) {
return this;
}
+ public WriteBuilder lazy() {
+ this.lazyWriter = true;
+ return this;
+ }
+
+ public WriteBuilder lazy(boolean lazy) {
+ this.lazyWriter = lazy;
+ return this;
+ }
+
Review comment:
There are actually situations when you might want an empty file to exist as soon as you get ready to write.
Some of the tests also assume this behaviour.
I did not want to change the default behaviour in this regard and the builder can leave this to the API user.
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -187,11 +215,6 @@ private void flushRowGroup(boolean finished) {
private void startRowGroup() {
Preconditions.checkState(!closed, "Writer is closed");
- try {
- this.nextRowGroupSize = Math.min(writer.getNextRowGroupSize(), targetRowGroupSize);
- } catch (IOException e) {
- throw new RuntimeIOException(e);
- }
Review comment:
Wasn't sure if I made sense, hope that cleared it up :)
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r742130858
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
Sorry, I'm not sure I understand what's happening. Does the test fail? What is the complete error message?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus edited a comment on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus edited a comment on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-947959498
@rdblue right, that's what I figured. Does that not mean the entire test is pointless now? The folder/files will not exist before or after calling `abort` since the (now) lazy writer never even created them. The file count will always be 0 and the folder will never exist.
I can change the test to ensure that but it won't really check any of `abort`'s behaviour, will it?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r742173494
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
I am not sure how the Flink partitioning logic works. Without this, it does not create enough files and fails the test
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r735142006
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
Can you elaborate more? What are you seeing when you run this without the change?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r735538819
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
The partitioning logic seems to only create half the files it is meant to without 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus closed pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus closed pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-971768810
@NeQuissimus, I think we'd love to get this in. The blocker is that test. We need to understand what's going on and fix it. I can't say when I'll have time to investigate myself, so if you want to find out more about it that would be really helpful!
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r756456476
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -139,9 +150,9 @@ public Metrics metrics() {
public long length() {
try {
if (closed) {
- return writer.getPos();
+ return writer().getPos();
Review comment:
Testing suggestion: It seems that possibly another flag might be needed of whether or not the writer has been instantiated at all and then throw an exception if `getPos` is called on a not open writer or in other situations. That’s likely what’s needed when debugging the tests (and other places that `writer()` has been swapped for `writer`). And similarly with the calls for the footer.
I think that between Flink tasks and operators, somehow across checkpoints it’s not preserving the instantiated writer (and underlying buffer), which would have been the previous behavior.
Or for testing possibly forcing flushing based on some calls that are now silently re-instantiating the writer via just calling `writer()` where previously there was a hard created writer (and it’s used to determine if the file should write).
Particularly the mixing of `writer()` with the `writeStore` is somewhat suspicious to me in the context of Flink’s task and operator isolated state and how it gets serialized across the cluster.
But I think you need to ensure the data in the parquet writer gets buffered for Flink.
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
##########
@@ -139,9 +150,9 @@ public Metrics metrics() {
public long length() {
try {
if (closed) {
- return writer.getPos();
+ return writer().getPos();
} else {
- return writer.getPos() + (writeStore.isColumnFlushNeeded() ? writeStore.getBufferedSize() : 0);
+ return writer().getPos() + (writeStore.isColumnFlushNeeded() ? writeStore.getBufferedSize() : 0);
Review comment:
This line is what I suspect is causing issues in Flink in particular.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-946973951
@rdblue Could you take a look at the test failures with me?
Both Flink tests that run `abort` use this code: https://github.com/apache/iceberg/blob/3763952c7b5d101e2c84efb1651b88439f641099/flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java#L226
With the writer here being lazy, it will not create the files the tests check for. And the actual writer is hidden away. I am unsure of the consequences, just a test with an invalid assumption?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-1000330564
Superseded by #3780
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-970559886
FWIW, we have been running this change and seen significant resource reduction.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#issuecomment-947869654
@NeQuissimus, the problem is that the data directory is no longer created in the test. I think what's happening is that the Parquet writer no longer creates an empty file that later gets deleted. That file create is what creates the data directory, so when we go to list it and check that no files were created, the check fails. You should be able to update the test code to check whether the directory exists and handle this case.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r735018022
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
Why is this change needed?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r742173494
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
I am not sure how the Flink partitioning logic works. Without this, it does not create enough files and fails the test
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] NeQuissimus commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
NeQuissimus commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r742173494
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
I am not sure how the Flink partitioning logic works. Without this, it does not create enough files and fails the test
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
I am not sure how the Flink partitioning logic works. Without this, it does not create enough files and fails the test
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3293: Lazily initialize the writer inside ParquetWriter
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3293:
URL: https://github.com/apache/iceberg/pull/3293#discussion_r742130858
##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java
##########
@@ -219,9 +219,15 @@ private void testAbort(boolean partitioned) throws IOException {
writer.write(createUpdateBefore(1, "aaa"));
writer.write(createUpdateAfter(1, "bbb"));
+ if (partitioned) {
+ writer.complete();
+ }
Review comment:
Sorry, I'm not sure I understand what's happening. Does the test fail? What is the complete error message?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org