Core: Add new writer interfaces

core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java

@@ -19,7 +19,6 @@
 
 package org.apache.iceberg.deletes;
 
-import java.io.Closeable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import org.apache.iceberg.DeleteFile;
@@ -29,10 +28,12 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.DeleteWriteResult;
 import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.Writer;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
-public class EqualityDeleteWriter<T> implements Closeable {
+public class EqualityDeleteWriter<T> implements Writer<T, DeleteWriteResult> {
   private final FileAppender<T> appender;
   private final FileFormat format;
   private final String location;
@@ -56,10 +57,17 @@ public EqualityDeleteWriter(FileAppender<T> appender, FileFormat format, String
     this.equalityFieldIds = equalityFieldIds;
   }
 
+  @Override
+  public void write(T row) throws IOException {
+    appender.add(row);
+  }
+
+  @Deprecated
   public void deleteAll(Iterable<T> rows) {
     appender.addAll(rows);
   }
 
+  @Deprecated
   public void delete(T row) {
     appender.add(row);
   }
@@ -89,4 +97,9 @@ public DeleteFile toDeleteFile() {
     Preconditions.checkState(deleteFile != null, "Cannot create delete file from unclosed writer");
     return deleteFile;
   }
+
+  @Override
+  public DeleteWriteResult result() {
+    return new DeleteWriteResult(toDeleteFile());
+  }
 }

core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java

@@ -19,7 +19,6 @@
 
 package org.apache.iceberg.deletes;
 
-import java.io.Closeable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import org.apache.iceberg.DeleteFile;
@@ -28,11 +27,13 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.DeleteWriteResult;
 import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.Writer;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.CharSequenceSet;
 
-public class PositionDeleteWriter<T> implements Closeable {
+public class PositionDeleteWriter<T> implements Writer<PositionDelete<T>, DeleteWriteResult> {
   private final FileAppender<StructLike> appender;
   private final FileFormat format;
   private final String location;
@@ -55,15 +56,27 @@ public PositionDeleteWriter(FileAppender<StructLike> appender, FileFormat format
     this.pathSet = CharSequenceSet.empty();
   }
 
+  @Override
+  public void write(PositionDelete<T> positionDelete) throws IOException {
+    pathSet.add(positionDelete.path());
+    appender.add(positionDelete);
+  }
+
+  @Deprecated
   public void delete(CharSequence path, long pos) {
     delete(path, pos, null);
   }
 
+  @Deprecated
   public void delete(CharSequence path, long pos, T row) {
     pathSet.add(path);
     appender.add(delete.set(path, pos, row));
   }
 
+  public long length() {
+    return appender.length();
+  }
+
   @Override
   public void close() throws IOException {
     if (deleteFile == null) {
@@ -88,4 +101,9 @@ public DeleteFile toDeleteFile() {
     Preconditions.checkState(deleteFile != null, "Cannot create delete file from unclosed writer");
     return deleteFile;
   }
+
+  @Override
+  public DeleteWriteResult result() {
+    return new DeleteWriteResult(toDeleteFile(), referencedDataFiles());
+  }
 }

core/src/main/java/org/apache/iceberg/io/BaseDeltaTaskWriteResult.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.io;
+
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public class BaseDeltaTaskWriteResult implements DeltaTaskWriter.Result {
+
+  private final DataFile[] dataFiles;
+  private final DeleteFile[] deleteFiles;
+  private final CharSequenceSet referencedDataFiles;
+
+  public BaseDeltaTaskWriteResult(List<DataFile> dataFiles, List<DeleteFile> deleteFiles,
+                                  CharSequenceSet referencedDataFiles) {
+    this.dataFiles = dataFiles.toArray(new DataFile[0]);
+    this.deleteFiles = deleteFiles.toArray(new DeleteFile[0]);
+    this.referencedDataFiles = referencedDataFiles;
+  }
+
+  @Override
+  public DataFile[] dataFiles() {
+    return dataFiles;
+  }
+
+  @Override
+  public DeleteFile[] deleteFiles() {
+    return deleteFiles;
+  }
+
+  @Override
+  public CharSequenceSet referencedDataFiles() {
+    return referencedDataFiles;
+  }
+}

core/src/main/java/org/apache/iceberg/io/BaseDeltaTaskWriter.java

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.io;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CharSequenceSet;
+import org.apache.iceberg.util.Tasks;
+
+public abstract class BaseDeltaTaskWriter<T> implements DeltaTaskWriter<T> {
+
+  private final List<DataFile> dataFiles = Lists.newArrayList();
+  private final List<DeleteFile> deleteFiles = Lists.newArrayList();
+  private final CharSequenceSet referencedDataFiles = CharSequenceSet.empty();
+  private final FileIO io;
+
+  private boolean closed = false;
+
+  public BaseDeltaTaskWriter(FileIO io) {
+    this.io = io;
+  }
+
+  protected FileIO io() {
+    return io;
+  }
+
+  @Override
+  public void abort() throws IOException {
+    Preconditions.checkState(closed, "Cannot abort unclosed task writer");
+
+    Tasks.foreach(Iterables.concat(dataFiles, deleteFiles))
+        .suppressFailureWhenFinished()
+        .noRetry()
+        .run(file -> io.deleteFile(file.path().toString()));
+  }
+
+  @Override
+  public Result result() {
+    Preconditions.checkState(closed, "Cannot obtain result from unclosed task writer");
+    return new BaseDeltaTaskWriteResult(dataFiles, deleteFiles, referencedDataFiles);
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (!closed) {
+      closeWriters();
+      this.closed = true;
+    }
+  }
+
+  protected abstract void closeWriters() throws IOException;
+
+  protected void closeDataWriter(PartitionAwareWriter<?, DataWriteResult> writer) throws IOException {
+    writer.close();
+
+    DataWriteResult result = writer.result();
+    dataFiles.addAll(result.dataFiles());
+  }
+
+  protected void closeDeleteWriter(PartitionAwareWriter<?, DeleteWriteResult> deleteWriter) throws IOException {
+    deleteWriter.close();
+
+    DeleteWriteResult result = deleteWriter.result();
+    deleteFiles.addAll(result.deleteFiles());
+    referencedDataFiles.addAll(result.referencedDataFiles());
+  }
+}

core/src/main/java/org/apache/iceberg/io/CDCTaskWriter.java

@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.io;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.util.StructLikeMap;
+import org.apache.iceberg.util.StructProjection;
+
+public class CDCTaskWriter<T> extends BaseDeltaTaskWriter<T> {
+
+  private final PartitionAwareWriter<T, DataWriteResult> dataWriter;
+  private final PartitionAwareWriter<T, DeleteWriteResult> equalityDeleteWriter;
+  private final PartitionAwareWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter;
+  private final StructProjection keyProjection;
+  private final Map<StructLike, PartitionAwarePathOffset> insertedRows;
+  private final PositionDelete<T> positionDelete;
+  private final Function<T, StructLike> toStructLike;
+
+  public CDCTaskWriter(PartitionAwareWriter<T, DataWriteResult> dataWriter,
+                       PartitionAwareWriter<T, DeleteWriteResult> equalityDeleteWriter,
+                       PartitionAwareWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
+                       FileIO io, Schema schema, Schema deleteSchema,
+                       Function<T, StructLike> toStructLike) {
+    super(io);
+    this.dataWriter = dataWriter;
+    this.equalityDeleteWriter = equalityDeleteWriter;
+    this.positionDeleteWriter = positionDeleteWriter;
+    this.positionDelete = new PositionDelete<>();
+    this.keyProjection = StructProjection.create(schema, deleteSchema);
+    this.insertedRows = StructLikeMap.create(deleteSchema.asStruct());
+    this.toStructLike = toStructLike;
+  }
+
+  @Override
+  public void insert(T row, PartitionSpec spec, StructLike partition) throws IOException {
+    CharSequence currentPath = dataWriter.currentPath(spec, partition);
+    long currentPosition = dataWriter.currentPosition(spec, partition);
+    PartitionAwarePathOffset offset = new PartitionAwarePathOffset(spec, partition, currentPath, currentPosition);
+
+    StructLike copiedKey = StructCopy.copy(keyProjection.wrap(toStructLike.apply(row)));
+
+    PartitionAwarePathOffset previous = insertedRows.put(copiedKey, offset);
+    if (previous != null) {
+      // TODO: attach the previous row if has a position delete row schema
+      positionDelete.set(previous.path(), previous.rowOffset(), null);
+      positionDeleteWriter.write(positionDelete, spec, partition);
+    }
+
+    dataWriter.write(row, spec, partition);
+  }
+
+  @Override
+  public void delete(T row, PartitionSpec spec, StructLike partition) throws IOException {
+    StructLike key = keyProjection.wrap(toStructLike.apply(row));
+    PartitionAwarePathOffset previous = insertedRows.remove(key);
+    if (previous != null) {
+      // TODO: attach the previous row if has a position delete row schema
+      positionDelete.set(previous.path(), previous.rowOffset(), null);
+      positionDeleteWriter.write(positionDelete, previous.spec, previous.partition);
+    }
+
+    equalityDeleteWriter.write(row, spec, partition);
+  }
+
+  @Override
+  public void delete(CharSequence path, long pos, T row, PartitionSpec spec, StructLike partition) throws IOException {
+    throw new IllegalArgumentException(this.getClass().getName() + " does not implement explicit position delete");
+  }
+
+  @Override
+  protected void closeWriters() throws IOException {
+    if (dataWriter != null) {
+      closeDataWriter(dataWriter);
+    }
+
+    if (equalityDeleteWriter != null) {
+      closeDeleteWriter(equalityDeleteWriter);
+    }
+
+    if (positionDeleteWriter != null) {
+      closeDeleteWriter(positionDeleteWriter);
+    }
+  }
+
+  private static class PartitionAwarePathOffset {
+    private final PartitionSpec spec;
+    private final StructLike partition;
+    private final CharSequence path;
+    private final long rowOffset;
+
+    private PartitionAwarePathOffset(PartitionSpec spec, StructLike partition, CharSequence path, long rowOffset) {
+      this.spec = spec;
+      this.partition = partition;
+      this.path = path;
+      this.rowOffset = rowOffset;
+    }
+
+    public PartitionSpec spec() {
+      return spec;
+    }
+
+    public StructLike partition() {
+      return partition;
+    }
+
+    public CharSequence path() {
+      return path;
+    }
+
+    public long rowOffset() {
+      return rowOffset;
+    }
+  }
+}