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/05/10 00:55:07 UTC
[GitHub] [iceberg] openinx opened a new issue #2571: Flink batch job cannot read the iceberg table that writing few chang-log events in format v2.
openinx opened a new issue #2571:
URL: https://github.com/apache/iceberg/issues/2571
I tried to add few lines code to read the iceberg table after writing few equality delete records in format v2 as following, It will encounter an NullPointerException. I think it's worth to take some time to dig in what's the cause.
```patch
diff --git a/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
index fd2a71ab..dfe4ec24 100644
--- a/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
+++ b/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
@@ -44,6 +45,7 @@ import org.apache.iceberg.flink.MiniClusterResource;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.flink.TestTableLoader;
import org.apache.iceberg.flink.source.BoundedTestSource;
+import org.apache.iceberg.flink.source.FlinkSource;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -162,6 +164,16 @@ public class TestFlinkIcebergSinkV2 extends TableTestBase {
// Execute the program.
env.execute("Test Iceberg Change-Log DataStream.");
+ DataStream<RowData> batch = FlinkSource.forRowData()
+ .env(env)
+ .tableLoader(tableLoader)
+ .streaming(false)
+ .build();
+
+ batch.print();
+
+ env.execute("Test Iceberg Change-Log Batch Read.");
+
table.refresh();
List<Snapshot> snapshots = findValidSnapshots(table);
int expectedSnapshotNum = expectedRecordsPerCheckpoint.size();
```
```txt
[flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: Custom Source -> Sink: Print to Std. Out (1/1) (1f110148951a8e6b9a131546ceffee9e) switched from RUNNING to FAILED on b6da4e08-c07d-4b4c-829d-5d76237be407 @ localhost (dataPort=-1).
java.lang.NullPointerException
at org.apache.iceberg.data.DeleteFilter.pos(DeleteFilter.java:108)
at org.apache.iceberg.deletes.Deletes$PositionSetDeleteFilter.shouldKeep(Deletes.java:157)
at org.apache.iceberg.util.Filter$Iterator.shouldKeep(Filter.java:49)
at org.apache.iceberg.io.FilterIterator.advance(FilterIterator.java:67)
at org.apache.iceberg.io.FilterIterator.hasNext(FilterIterator.java:50)
at org.apache.iceberg.io.FilterIterator.advance(FilterIterator.java:65)
at org.apache.iceberg.io.FilterIterator.hasNext(FilterIterator.java:50)
at org.apache.iceberg.flink.source.DataIterator.updateCurrentIterator(DataIterator.java:100)
at org.apache.iceberg.flink.source.DataIterator.hasNext(DataIterator.java:84)
at org.apache.iceberg.flink.source.FlinkInputFormat.reachedEnd(FlinkInputFormat.java:104)
at org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:89)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:241)
```
--
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] chenjunjiedada commented on issue #2571: Flink batch job cannot read the iceberg table that writing few chang-log events in format v2.
Posted by GitBox <gi...@apache.org>.
chenjunjiedada commented on issue #2571:
URL: https://github.com/apache/iceberg/issues/2571#issuecomment-836055150
Did you write positional delete as well? From the code path, it looks like the `posDeletes` is not equal to null. The `fileProjection` only project `_pos` when `posDeletes` is not empty.
--
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] wulei0302 commented on issue #2571: Flink batch job cannot read the iceberg table that writing few chang-log events in format v2.
Posted by GitBox <gi...@apache.org>.
wulei0302 commented on issue #2571:
URL: https://github.com/apache/iceberg/issues/2571#issuecomment-839536993
I think this may be a problem with FlinkAvroReader. When reading the avro data file, FlinkAvroReader did not return the position information of the record, which caused the NEP problem later. Can we refer to FlinkParquetReaders ?
--
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org