@Override
        public void close() throws IOException {
            try {
                // Close data writer and add completed data files.
                if (dataWriter != null) {
                    try {
                        dataWriter.close();
                    } finally {
                        dataWriter = null;
                    }
                }

                // Close eq-delete writer and add completed equality-delete files.
                if (eqDeleteWriter != null) {
                    try {
                        eqDeleteWriter.close();
                    } finally {
                        eqDeleteWriter = null;
                    }
                }

                if (insertedRowMap != null) {
                    insertedRowMap.clear();
                    insertedRowMap = null;
                }

                // Add the completed pos-delete files.
                if (posDeleteWriter != null) {
                    try {
                        // complete will call close
                        completedDeleteFiles.addAll(posDeleteWriter.complete());
                        referencedDataFiles.addAll(posDeleteWriter.referencedDataFiles());
                    } finally {
                        posDeleteWriter = null;
                    }
                }
            } catch (IOException | RuntimeException e) {
                setFailure(e);
                throw e;
            }
        }
Delete those rows with the given key. It will only write the values of equality fields into
the equality-delete file.

@param key is the projected data whose columns are the same as the equality fields.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/trick/BaseTaskWriter.java