public void write(T row) throws IOException {
            BaseTaskWriter.PathOffset pathOffset =
                    BaseTaskWriter.PathOffset.of(dataWriter.currentPath(), dataWriter.currentRows());

            // Create a copied key from this row.
            StructLike copiedKey = StructCopy.copy(structProjection.wrap(asStructLike(row)));

            // Adding a pos-delete to replace the old path-offset.
            BaseTaskWriter.PathOffset previous = insertedRowMap.put(copiedKey, pathOffset);
            if (previous != null) {
                // TODO attach the previous row if has a positional-delete row schema in appender factory.
                posDeleteWriter.delete(previous.path, previous.rowOffset, null);
            }

            dataWriter.write(row);
        }
Wrap the passed in key of a row as a {@link StructLike}
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/trick/BaseTaskWriter.java