Andreas,
I think the problem is that you're using the IDs from the schema you used
to create the delete table, not the IDs from the data table itself. If I'm
right, you should be able to fix this by changing how you are setting
`equalityFieldIds` to use `dataTable.schema().select(...)` instead of
`icebergSchema`.
When a table is created, Iceberg reassigns all of the IDs in the incoming
schema to make sure they're consistent. So if you've created a schema using
IDs other than the ones that would be automatically assigned (for example,
starting at 0) then the IDs will be different. That would explain the
failure at DeleteFileIndex.java:193, which is looking up the equality field
IDs in the table schema. I'm not sure how you've created these schemas or
what their relationships are, though, so I'm mostly speculating.
The important thing is that you always want to use a table's schema rather
than the schema you used to create the table or another table's schema.
Ryan
On Wed, Apr 20, 2022 at 5:50 PM Hailu, Andreas <andreas.ha...@gs.com> wrote:
> Hi folks,
>
>
>
> I’m trying to test out record deletions on a V2 table but I seem to be
> corrupting the Table and receive a NullPointerException trying to read the
> Table post deletion.
>
>
>
> I have a main Table ‘T1’ containing 2 records, and another Table ‘T2’ that
> contains 1 record from T1 that I want to delete. I create a DeleteFile out
> of T2 using the EqualityDeleteWriter, and then apply the RowDelta to T1
> with this DeleteFile. Post this sequence, this exception is thrown trying
> to read:
>
>
>
> Caused by: java.lang.NullPointerException
>
> at
> org.apache.iceberg.DeleteFileIndex.canContainEqDeletesForFile(DeleteFileIndex.java:193)
>
> at
> org.apache.iceberg.DeleteFileIndex.canContainDeletesForFile(DeleteFileIndex.java:144)
>
> at
> org.apache.iceberg.DeleteFileIndex.lambda$forDataFile$3(DeleteFileIndex.java:134)
>
> at
> java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:174)
>
> at
> java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
>
> at
> java.util.stream.Streams$ConcatSpliterator.forEachRemaining(Streams.java:742)
>
> at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
>
> at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
>
> at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
>
> at
> java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
>
> at
> java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
>
> at
> org.apache.iceberg.DeleteFileIndex.forDataFile(DeleteFileIndex.java:135)
>
> at
> org.apache.iceberg.DeleteFileIndex.forEntry(DeleteFileIndex.java:115)
>
> at
> org.apache.iceberg.ManifestGroup.lambda$planFiles$6(ManifestGroup.java:181)
>
> at
> org.apache.iceberg.io.CloseableIterable$4$1.next(CloseableIterable.java:113)
>
> at
> org.apache.iceberg.io.CloseableIterable$ConcatCloseableIterable$ConcatCloseableIterator.next(CloseableIterable.java:206)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.TransformedIterator.next(TransformedIterator.java:47)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.TransformedIterator.next(TransformedIterator.java:47)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.Iterators$ConcatenatedIterator.hasNext(Iterators.java:1330)
>
> at
> org.apache.iceberg.io.CloseableIterator$1.hasNext(CloseableIterator.java:50)
>
> at
> org.apache.iceberg.util.BinPacking$PackingIterator.hasNext(BinPacking.java:106)
>
> at
> org.apache.iceberg.io.CloseableIterator$1.hasNext(CloseableIterator.java:50)
>
> at
> org.apache.iceberg.io.CloseableIterable$4$1.hasNext(CloseableIterable.java:108)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.TransformedIterator.hasNext(TransformedIterator.java:42)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.TransformedIterator.hasNext(TransformedIterator.java:42)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.Iterators$ConcatenatedIterator.getTopMetaIterator(Iterators.java:1309)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.Iterators$ConcatenatedIterator.hasNext(Iterators.java:1325)
>
> at
> org.apache.iceberg.relocated.com.google.common.collect.TransformedIterator.hasNext(TransformedIterator.java:42)
>
> at
> org.apache.iceberg.io.CloseableIterable.concat(CloseableIterable.java:122)
>
> at
> org.apache.iceberg.data.GenericReader.open(GenericReader.java:65)
>
> at
> org.apache.iceberg.data.TableScanIterable.iterator(TableScanIterable.java:41)
>
> at
> com.gs.ep.da.lake.refinerlib.flink.iceberg.poc.IcebergApplyDeltas.logRecordsInTable(IcebergApplyDeltas.java:111)
>
>
>
> I’m reading the table through IcebergGenerics.read() API, which works fine
> prior to this. Is there something wrong with this approach? I’ve included
> my code snippet below.
>
>
>
> // ...
>
> Table deleteTable = catalog.createTable(deleteDeltaTableIdentifier,
> icebergSchema, null, properties);
>
> DataStream<GenericRecord> recordsForDelete =
> readParquet(getStreamExecutionEnvironment(), avroSchema,
> deltaDeletesLocation);
>
> writeToTable(recordsForDelete, avroSchema, icebergSchema,
> getTableLoader(deleteTable), deleteTable); // writeToTable() writes
> GenericRecords to Table through Iceberg Flink API
>
> deleteTable = catalog.loadTable(deleteDeltaTableIdentifier); // This Table
> contains single record looking to delete from dataTable initialized later
>
>
>
> TableIdentifier dataTableName =
> TableIdentifier.of(icebergProofOfConceptNamespace, getIcebergTableName());
>
> Table dataTable = catalog.loadTable(dataTableName); // Main Table
> consiting of 2 records, 1 to be deleted
>
> logger.info("Logging records before update");
>
> logRecordsInTable(dataTable); // Works fine
>
>
>
> logger.info("Logging records for deletion");
>
> logRecordsInTable(deleteTable); // Works fine
>
> OutputFileFactory fileFactory = OutputFileFactory.builderFor(dataTable, 1,
> 1).build();
>
> OutputFile deleteOutputFile =
> fileFactory.newOutputFile().encryptingOutputFile();
>
> logger.info("Performing deletes");
>
> EqualityDeleteWriter<Record> deleteWriter =
> Parquet.writeDeletes(deleteOutputFile)
>
> .forTable(dataTable)
>
> .createWriterFunc(GenericParquetWriter::buildWriter)
>
> .overwrite()
>
>
> .equalityFieldIds(icebergSchema.columns().stream().mapToInt(Types.NestedField::fieldId).toArray())
>
> .buildEqualityWriter();
>
> try(Closeable closeable = deleteWriter; CloseableIterable<Record> deletes
> = IcebergGenerics.read(deleteTable).build()) {
>
> deleteWriter.write(deletes);
>
> }
>
> DeleteFile deletes = deleteWriter.toDeleteFile();
>
> Transaction deltaTransaction = dataTable.newTransaction();
>
> deltaTransaction.newRowDelta()
>
> .addDeletes(deletes)
>
> .commit();
>
> deltaTransaction.commitTransaction();
>
>
>
> logger.info("Transaction complete");
>
> logger.info("Logging records post update");
>
> logRecordsInTable(dataTable); // Results in NPE
>
> // ...
>
>
>
> private void logRecordsInTable(Table dataTable) throws IOException {
>
> try(CloseableIterable<Record> postDeletionRes =
> IcebergGenerics.read(dataTable).build()) {
>
> List<String> columnNames =
> dataTable.schema().columns().stream().map(Types.NestedField::name).collect(Collectors.toList());
>
> for(Record res : postDeletionRes) {
>
> StringBuilder sb = new StringBuilder().append("[ ");
>
> columnNames.forEach(name -> {
>
> sb.append(name)
>
> .append(": ")
>
> .append(res.getField(name))
>
> .append(", ");
>
> });
>
>
>
> sb.append("]");
>
> logger.info(sb.toString());
>
> }
>
> }
>
> }
>
>
>
>
>
> best,
>
> ah
>
>
>
> ------------------------------
>
> Your Personal Data: We may collect and process information about you that
> may be subject to data protection laws. For more information about how we
> use and disclose your personal data, how we protect your information, our
> legal basis to use your information, your rights and who you can contact,
> please refer to: www.gs.com/privacy-notices
>
--
Ryan Blue
Tabular
