Core: Add new writer interfaces #2945

aokolnychyi · 2021-08-05T23:44:52Z

This PR has the following contributions:

Support for writing to multiple specs.
Support for StructLike in partitioned writers. Previously, we always assumed we can derive PartitionKey from data. That may not be the case for some operations as partitions may come from a metadata column.
A new DeltaWriter interface for writing data and deletes.
CDC writer that keeps track of records across partitions and can delete a key inserted in any partition.
More types of delete writers (e.g. clustered/fanout equality/position deletes).
Composition over inheritance to simplify reuse of writers.

New interfaces are added to core with an example of how they can be consumed in Spark 3.

Writer

The first major proposed API is the Writer interface that defines a contract for writing a number of files of a single type within one spec/partition. Existing DataWriter, EqualityDeleteWriter, PositionDeleteWriter classes are the simplest implementations of that API.

Then we have RollingWriter that implements Writer and wraps another writer to split the incoming records into multiple files within one spec/partition. We have RollingDataWriter, RollingEqualityDeleteWriter, RollingPositionDeleteWriter as actual implementations.

PartitionAwareWriter

All Writer implementations are limited to writing to a single spec/partition. To support writes to multiple specs and partitions, we have PartitionAwareWriter. In Iceberg, we support two types of writes: fanout and clustered. That’s why I am proposing to add ClusteredWriter and FanoutWriter. On one hand, ClusteredWriter will write to multiple specs and partitions ensuring the incoming data is properly clustered. On the other hand, FanoutWriter will keep a number of writers open and will not require a particular order of data. ClusteredWriter is very similar to our existing PartitionedWriter but it also detects changes in the spec, not only in partition values.

DeltaTaskWriter

This PR also introduces a new DeltaTaskWriter interface that will be used by query engine integrations.

aokolnychyi · 2021-08-05T23:52:49Z

cc @openinx @stevenzwu @RussellSpitzer @rdblue @cwsteinbach @danielcweeks @kbendick @karuppayya @flyrain @pvary @jackye1995 @yyanyy @szehon-ho @rymurr @jun-he

This PR is still work in progress but I'd like to get some early feedback. I've benchmarked new writers locally using async-profiler. I'll clean the benchmarking code later and push here. Once we agree on the APIs, I'll split the PR into smaller chunks and add tests.

As it is a large change, I'd advise to start with Writer, then PartitionAwareWriter, then TaskWriter and DeltaTaskWriter.

aokolnychyi · 2021-08-05T23:55:34Z

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

+import org.apache.iceberg.util.StructLikeMap;
+import org.apache.iceberg.util.StructProjection;
+
+public class CDCTaskWriter<T> extends BaseDeltaTaskWriter<T> {


This is essentially a copy of the existing BaseEqualityDeltaWriter but using new abstractions.

aokolnychyi · 2021-08-05T23:56:19Z

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

+/**
+ * A writer capable of writing to multiple specs and partitions ensuring the incoming records are properly clustered.
+ */
+public abstract class ClusteredWriter<T, R> implements PartitionAwareWriter<T, R> {


Naming suggestions are welcome.

I like "clustered" since that's the assumption.

The naming has a slight conflict with the "sorted" position delete writer. That writer keeps deletes in memory and sorts them prior to writing them out. That's for the CDC use case where position deletes may be in any order, rather than the MERGE use case where we should be able to ask the engine to produce deletes in the expected file/pos order.

Yeah, any ideas here? The main distinction is that ClusteredWriter writes to multiple specs/partitions and just checks the data is clustered. That writer actually sorts position deletes and writes only to a single spec/partition.

aokolnychyi · 2021-08-05T23:57:07Z

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

+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * A writer capable of writing to multiple specs and partitions ensuring the incoming records are properly clustered.


This one is similar to our existing PartitionedWriter but with some notable differences I'll mention below.

aokolnychyi · 2021-08-05T23:58:40Z

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

+
+  @Override
+  public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+    if (!spec.equals(currentSpec)) {


We support writing across multiple specs now. Background here.

aokolnychyi · 2021-08-05T23:59:56Z

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

+  protected abstract R aggregatedResult();
+
+  @Override
+  public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {


Instead of deriving a partition struct through PartitionKey, the new writer accepts StructLike. In some cases, this will come from a metadata column instead.

aokolnychyi · 2021-08-06T00:02:49Z

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

+      currentPartition = partition != null ? StructCopy.copy(partition) : null;
+      currentWriter = newWriter(currentSpec, currentPartition);
+
+    } else if (partition != currentPartition && partitionComparator.compare(partition, currentPartition) != 0) {


This place is important as it is invoked for every single row. Previously, we used equals in PartitionKey. In this PR, I am using our struct comparator as the passed StructLike may not necessarily be PartitionKey.

My benchmarks show this is on par with the previous implementation.

aokolnychyi · 2021-08-06T00:03:54Z

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

+      }
+
+      // copy the partition key as the key object is reused
+      currentPartition = partition != null ? StructCopy.copy(partition) : null;


Using StructCopy instead of PartitionKey#copy for the same reasons as above. This is less critical and is invoked only when we detect a partition change (i.e. NOT for every row).

aokolnychyi · 2021-08-06T00:05:14Z

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

+
+      currentSpec = spec;
+      partitionComparator = Comparators.forType(partitionType);
+      completedPartitions = StructLikeSet.create(partitionType);


Using StructLikeSet instead of a regular Set as we may get arbitrary StructLike implementations. This set is not going to be checked for every row so I don't worry a lot about performance here.

I think the original relied on PartitionKey to handle hashing and so it was safe. That was before we introduced StructLikeSet, which is probably the easier way to go.

Using PartitionKey is not even an option. For delete files, the partition struct will come from a metadata column, not by applying PartitionKey on data. It seems StructLikeSet is a nice solution and since we are not calling it for every single row, it should not be much more expensive.

aokolnychyi · 2021-08-06T00:06:16Z

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

+  private Writer<T, R> writer(PartitionSpec spec, StructLike partition) {
+    Map<StructLike, Writer<T, R>> specWriters = writers.computeIfAbsent(
+        spec.specId(),
+        id -> StructLikeMap.create(spec.partitionType()));


Using StructLikeMap here.

aokolnychyi · 2021-08-06T00:06:34Z

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

+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaTaskWriter<T> extends BaseDeltaTaskWriter<T> {


Naming TBD.

This is what we are going to use in Spark to write deltas during MERGE INTO.

aokolnychyi · 2021-08-06T00:08:42Z

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

+/**
+ * A writer capable of writing files of a single type (i.e. data/delete) to multiple specs and partitions.
+ */
+public interface PartitionAwareWriter<T, R> extends Closeable {


Initially, I made this extend Writer. However, that required introducing PartitionAwareRow that would wrap a row, spec, partition. Benchmarks showed that we spend some extra time wrapping every single row. I think the performance is more important in this case.

aokolnychyi · 2021-08-06T00:09:17Z

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

+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+public abstract class RollingWriter<T, W extends Writer<T, R>, R> implements Writer<T, R> {


Similar to our existing BaseRollingWriter.

aokolnychyi · 2021-08-06T00:10:46Z

spark3/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java

      PartitionSpec spec = table.spec();
      FileIO io = table.io();

+      OutputFileFactory fileFactory = OutputFileFactory.builderFor(table, partitionId, taskId)


This is an example of how to use the new writers in Spark 3.

aokolnychyi · 2021-08-06T00:11:39Z

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

+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {


We have V2 in the name temporarily. We already have TaskWriter and I did not want to change that in this PR.

Maybe, we better call it something else and keep the old hierarchy for now.

aokolnychyi · 2021-08-06T21:21:48Z

Here are some benchmark numbers for writing 2.5 million records (flat schema, 7 columns). I am using bucketing with 32 buckets on an int column for partitioned writes.

Benchmark                                                                       Mode  Cnt   Score   Error  Units
TaskWriterParquetBenchmark.writePartitionedDataNewFanoutWriter                    ss    5  10.432 ± 0.382   s/op
TaskWriterParquetBenchmark.writePartitionedDataOldFanoutWriter                    ss    5  11.315 ± 0.345   s/op
TaskWriterParquetBenchmark.writePartitionedDataNewWriter                          ss    5  11.416 ± 0.994   s/op
TaskWriterParquetBenchmark.writePartitionedDataOldWriter                          ss    5  11.331 ± 0.238   s/op
TaskWriterParquetBenchmark.writePartitionedEqualityDeleteNewWriter                ss    5  11.795 ± 1.553   s/op
TaskWriterParquetBenchmark.writeUnpartitionedDataNewWriter                        ss    5  10.736 ± 1.058   s/op
TaskWriterParquetBenchmark.writeUnpartitionedDataOldWriter                        ss    5  10.501 ± 2.084   s/op
TaskWriterParquetBenchmark.writeUnpartitionedEqualityDeleteNewWriter              ss    5   9.935 ± 0.166   s/op
TaskWriterParquetBenchmark.writeUnpartitionedPositionDeleteWithoutRowNewWriter    ss    5   8.833 ± 0.791   s/op

Memory-wise it is very similar. Here is an example.

TaskWriterParquetBenchmark.writePartitionedDataNewWriter:·gc.alloc.rate                                            ss    5         177.302 ±        17.914  MB/sec
TaskWriterParquetBenchmark.writePartitionedDataNewWriter:·gc.churn.G1_Eden_Space                                   ss    5         136.865 ±        12.818  MB/sec
TaskWriterParquetBenchmark.writePartitionedDataNewWriter:·gc.churn.G1_Old_Gen                                      ss    5           5.411 ±         0.646  MB/sec
TaskWriterParquetBenchmark.writePartitionedDataOldWriter:·gc.alloc.rate                                            ss    5         177.730 ±        11.985  MB/sec
TaskWriterParquetBenchmark.writePartitionedDataOldWriter:·gc.churn.G1_Eden_Space                                   ss    5         137.768 ±        21.407  MB/sec
TaskWriterParquetBenchmark.writePartitionedDataOldWriter:·gc.churn.G1_Old_Gen                                      ss    5           5.420 ±         0.892  MB/sec

aokolnychyi · 2021-08-06T22:45:44Z

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

+  @Override
+  protected void add(DeleteWriteResult result) {
+    deleteFiles.addAll(result.deleteFiles());
+    referencedDataFiles.addAll(result.referencedDataFiles());


I noticed CharSequenceSet#addAll calls in flame graphs (nothing too bad). If we want to squeeze a little bit of performance, we can replace addAll with union to avoid coping things. Alternatively, we can detect whether the set we add is also CharSequenceSet. If so, no need to unwrap and wrap records again.

I am not sure it is going to be worth it, though.

aokolnychyi · 2021-08-06T23:06:31Z

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

+  void delete(CharSequence path, long pos, T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+  // position delete without persisting row
+  default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike partition) throws IOException {


One potential optimization is to wrap query engine specific String representations as CharSequence. For example, we do spend some noticeable amount of time converting UTF8String into Java String whenever we retrieve the path from InternalRow. Should be investigated separately, though.

Agreed. UTF8String handling is something we should probably look into independently.

szehon-ho

Looks ok to me from my limited experience of Writers.

A bit of a bummer about not able to do inheritance, but saw that it's about performance and not having to wrap the row and partitionSpec

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

rdblue · 2021-08-09T20:35:32Z

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

+
+  void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+  void abort() throws IOException;


What is the expectation for calling abort? Should the writer already be closed?

This essentially mimics the current interface we have. I am happy to reconsider having abort in our task writers. I guess the idea is to reuse the cleanup logic in all query engine integrations. We could have some utility methods instead.

I removed abort and it looks like we don't really need V2TaskWriter anymore. So I removed that interface completely and just kept DeltaWriter.

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

rdblue · 2021-08-09T20:41:11Z

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

  }

+  @Override
+  public void write(T row) throws IOException {


Do we need to throw IOException if the current writer does not?

Probably, no. I guess IOException was just defined in the parent interface.

openinx · 2021-08-26T02:04:02Z

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

+    appender.add(row);
+  }
+
+  @Deprecated


We don't expose those deleteAll and delete methods in the public iceberg-api module, so is there any neccessary to keep those deprecated API for at least a release ? I'm thinking that we could just remove those from this class.

I'll be up for that. I did not do this in this WIP PR to avoid touching more places. I think this is a low-level API which we can break.

It's a low-level API, but it could be used by external projects like Hive since this is the easiest way to correctly write Iceberg files. I think we should deprecate them like Anton did here.

openinx · 2021-08-26T02:14:53Z

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


+  @Override
+  public void write(PositionDelete<T> positionDelete) throws IOException {
+    pathSet.add(positionDelete.path());


Nit: I think it's clear to rename pathSet as referencedDataFiles. When I check this variable at the first glance, I was thinking: pathSet ? which path set ? what's used for. I did not get it until I checked the referencedDataFiles() method.

I agree. I'll do that rename when I split this into smaller PRs.

+1 for the rename.

openinx · 2021-09-01T12:17:02Z

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

+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaWriter<T> extends BaseDeltaWriter<T> {


What's the specific case that MixedDeltaWriter will be used for ? If it's the batch UPDATE/DELETE case, then we should don't produce the equality-deletes , then I think we could just remain the delete(T row, PartitionSpec spec, StructLike partition) as Unsupported, we don't even wanna to initialize the equalityDeleteWriter.

openinx · 2021-09-01T12:45:50Z

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

+    Map<StructLike, FileWriter<T, R>> specWriters = writers.computeIfAbsent(
+        spec.specId(),
+        id -> StructLikeMap.create(spec.partitionType()));
+    FileWriter<T, R> writer = specWriters.get(partition);
+
+    if (writer == null) {
+      // copy the partition key as the key object is reused
+      StructLike copiedPartition = partition != null ? StructCopy.copy(partition) : null;
+      writer = newWriter(spec, copiedPartition);
+      specWriters.put(copiedPartition, writer);
+    }
+
+    return writer;
+  }


Could we maintain those writers from different partition spec and different partition into a flatten Map<K, FileWriter<T,R>> rather than a nested Map, by using a composited key <partitionSpecId, partitionData> . I think that will make the writer choose strategy quite simple, it will also simplify the ClusteredFileWriter's write method.

I can change that. No preference from my side here.

openinx · 2021-09-01T13:10:12Z

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

+/**
+ * A rolling equality delete writer that splits incoming deletes into multiple files within one spec/partition.
+ */
+public class RollingEqualityDeleteWriter<T> extends RollingDeleteWriter<T, EqualityDeleteWriter<T>> {


Is there necessary to share the abstracted RollingDeleteWriter between RollingEqualityDeleteWriter & RollingPositionDeleteWriter ? The key common thing is the referencedDataFiles I think, but the RollingEqualityDeleteWriter usually don't produce any referenced data files, right ? So I think we may could just make the RollingEqualityDeleteWriter extend the RollingFileWriter.

The idea was to avoid implementing addResult and aggregatedResult in both but now that I look at it, it is probably not worth a separate class. I'll remove it.

rdblue · 2021-09-17T19:31:57Z

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

+
+  @Override
+  protected DataWriter<T> newWriter(EncryptedOutputFile file) {
+    return writerFactory.newDataWriter(file, spec(), partition());


The current BaseRollingWriter implementation passes the partition key to this method each time. It looks like the new classes expose spec() and partition() to avoid passing them here, since these 3 method implementations are the only place where those getters are called. I think I'd prefer dropping the 2 getter methods and just passing the spec and partition in here. What do you think?

Alternatively, these implementations could keep their own copies of spec and partition in the constructor to avoid it. But I'd rather not do that.

I did not pass the spec and partition on purpose as our rolling writers, as opposed to other more sophisticated writers, can write only to a single partition. If I see the spec and partition being passed, I won't know it is always the same value.

I don't have a strong opinion here. If you feel it is not worth it, I can surely adapt.

rdblue · 2021-09-17T19:34:42Z

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

+    if (partition == null) {
+      this.currentFile = fileFactory.newOutputFile();
+    } else {
+      this.currentFile = fileFactory.newOutputFile(partition);


This should call newOutputFile(spec, partition), not this one that uses the default spec.

In fact, this version should probably be deprecated so that we can remove it once we move over to these writers.

Good catch, I think I did this PR before we had the new method. I'll update.

rdblue · 2021-09-17T19:37:29Z

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

+
+  private boolean shouldRollToNewFile() {
+    // TODO: ORC file now not support target file size before closed
+    return !fileFormat.equals(FileFormat.ORC) &&


With the new structure, does this still make sense? The file format is only passed through to this point for this check, but couldn't we just use a check before creating this class?

if (fileFormat == FileFormat.ORC) { return new DataWriter(...); } else { return new RollingDataWriter(...); }

(Not urgent to fix in this PR)

I remember I tried and then reverted. I look at it now and don't see why it did not work before. I'll update.

rdblue · 2021-09-17T19:40:14Z

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

+        io.deleteFile(currentFile.encryptingOutputFile());
+      } else {
+        R result = currentWriter.result();
+        addResult(result);


Minor: do we need result or can we just call addResult(currentWriter.result())?

rdblue · 2021-09-17T19:50:11Z

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

+  @Override
+  public void write(PositionDelete<T> positionDelete) throws IOException {
+    pathSet.add(positionDelete.path());
+    appender.add(positionDelete);


This class makes an assumption similar to the ClusteredFileWriter classes. Should we name it OrderedPositionDeleteWriter instead?

That will definitely be more descriptive. Do you think it will break anyone? This class existed for a while and we reference it in multiple places such as WriterFactory. I think either we keep the original name and deprecate old methods here and in EqualityDeleteWriter or just drop old methods and rename as needed.

rdblue · 2021-09-17T21:47:52Z

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

+      }
+
+      if (completedSpecs.contains(spec.specId())) {
+        throw new IllegalStateException("Already closed files for spec: " + spec.specId());


This doesn't need to be a requirement. We could pass (spec1, part1), (spec2, part3), (spec1, part2) and nothing would really break other than this. Do you think it is likely that clustering by spec is going to be the case most of the time? I'm not sure if this is something we should worry about removing.

I did this from the performance perspective to not keep a file open for each seen spec. Right now, I keep only a single file open at a time. I know we will cluster by spec in Spark merge-on-read. Do you have use cases in mind where we won't cluster by spec?

rdblue · 2021-09-17T21:48:50Z

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

+      partitionComparator = Comparators.forType(partitionType);
+      completedPartitions = StructLikeSet.create(partitionType);
+      // copy the partition key as the key object is reused
+      currentPartition = partition != null ? StructCopy.copy(partition) : null;


Can we move the null handling into StructCopy instead of doing it here?

rdblue · 2021-09-17T21:50:03Z

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

+      currentWriter.close();
+
+      R result = currentWriter.result();
+      addResult(result);


Useless variable, result?

rdblue · 2021-09-17T21:52:21Z

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

+
+  @Override
+  protected FileWriter<T, DataWriteResult> newWriter(PartitionSpec spec, StructLike partition) {
+    return new RollingDataWriter<>(writerFactory, fileFactory, io, fileFormat, targetFileSizeInBytes, spec, partition);


This is where it would be easy to do the check to see if fileFormat is ORC and skip creating a rolling writer if it is.

rdblue · 2021-09-17T21:59:07Z

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

+        id -> StructLikeMap.create(spec.partitionType()));
+    FileWriter<T, R> writer = specWriters.get(partition);
+
+    if (writer == null) {


Can we not use computeIfAbsent for the StructLikeMap as well?

Nevermind, it's the copy.

rdblue · 2021-09-17T23:16:54Z

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

+  void delete(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+  // position delete with persisting row
+  void delete(CharSequence path, long pos, T row, PartitionSpec spec, StructLike partition) throws IOException;


Other APIs put the optional row last. What about moving spec and partition before row? Would that be too inconsistent with others?

rdblue · 2021-09-17T23:22:49Z

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

+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaWriter<T> extends BaseDeltaWriter<T> {


I think this needs Javadoc and a better name that shows this only accepts position deletes.

rdblue · 2021-09-17T23:23:44Z

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

+import org.apache.iceberg.util.StructLikeMap;
+import org.apache.iceberg.util.StructProjection;
+
+public class CDCWriter<T> extends BaseDeltaWriter<T> {


This also needs some Javadoc to explain the context.

rdblue · 2021-09-17T23:34:32Z

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

+    this.dataWriter = dataWriter;
+    this.equalityDeleteWriter = equalityDeleteWriter;
+    this.positionDeleteWriter = positionDeleteWriter;
+    this.positionDelete = new PositionDelete<>();


Nit: PositionDelete.create()

The constructor should probably be made private.

rdblue · 2021-09-17T23:41:55Z

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

+
+  public CDCWriter(FanoutDataWriter<T> dataWriter,
+                   PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+                   PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,


I think that this needs to be FanoutSortedPositionDeleteWriter because the position deletes could be in any order.

rdblue · 2021-09-17T23:45:29Z

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

+      return path;
+    }
+
+    public long rowOffset() {


This is called position nearly everywhere else. Why call it rowOffset here?

rdblue · 2021-09-17T23:47:08Z

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

+    }
+  }
+
+  private static class PartitionAwarePathOffset {


Could this extend PositionDelete so it can be passed directly into the delete writer?

rdblue · 2021-09-17T23:48:56Z

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

+  }
+
+  @Override
+  public void delete(T row, PartitionSpec spec, StructLike partition) throws IOException {


I think we need to clearly document that this assumes that the row to delete has the same schema as the rows that will be inserted. We could also have a directDelete method that passes just the equality delete columns (key). That's worth considering if you want to split the interface for equality and position delete use cases.

rdblue · 2021-09-17T23:51:55Z

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

+    }
+  }
+
+  protected abstract void closeWriters() throws IOException;


This base class is really strange to me because the actual close is abstract, but then methods to close writers are in this implementation. I think I'd probably see if I could remove this class.

aokolnychyi · 2021-09-25T00:33:10Z

I am closing this one in favor of smaller PRs.

github-actions bot added core spark labels Aug 5, 2021

aokolnychyi force-pushed the v2-writers-proto-v3 branch from 6b1eff6 to 6a82e84 Compare August 5, 2021 23:50

aokolnychyi commented Aug 5, 2021

View reviewed changes

aokolnychyi commented Aug 6, 2021

View reviewed changes

Core: Add new writer interfaces

8fd5525

aokolnychyi force-pushed the v2-writers-proto-v3 branch from 6a82e84 to 8fd5525 Compare August 6, 2021 00:09

aokolnychyi commented Aug 6, 2021

View reviewed changes

rdblue requested review from openinx and rdblue August 6, 2021 15:22

aokolnychyi commented Aug 6, 2021

View reviewed changes

szehon-ho approved these changes Aug 7, 2021

View reviewed changes

rdblue reviewed Aug 9, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java Outdated Show resolved Hide resolved

rdblue reviewed Aug 9, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java Outdated Show resolved Hide resolved

rdblue reviewed Aug 9, 2021

View reviewed changes

openinx reviewed Sep 1, 2021

View reviewed changes

rdblue reviewed Sep 17, 2021

View reviewed changes

This was referenced Sep 18, 2021

Core: Add FileWriter interface #3149

Merged

Core: Add new rolling file writers #3158

Merged

Core: Add PartitioningWriter #3164

Merged

Core: Add position and equality delta writer interfaces #3176

Merged

aokolnychyi closed this Sep 25, 2021

dungdm93 mentioned this pull request Feb 15, 2022

[Core][Flink][Spark]: Refactor TaskWriter implementations #4132

Closed


		void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;

		void abort() throws IOException;

Core: Add new writer interfaces #2945

Core: Add new writer interfaces #2945

Uh oh!

Conversation

aokolnychyi commented Aug 5, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Writer

PartitionAwareWriter

DeltaTaskWriter

Uh oh!

aokolnychyi commented Aug 5, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aokolnychyi Aug 5, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aokolnychyi Aug 6, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aokolnychyi commented Aug 6, 2021

Uh oh!

aokolnychyi Aug 6, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aokolnychyi Aug 6, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aokolnychyi Aug 10, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aokolnychyi commented Aug 5, 2021 •

edited

Loading

aokolnychyi Aug 5, 2021 •

edited

Loading

aokolnychyi Aug 6, 2021 •

edited

Loading

aokolnychyi Aug 6, 2021 •

edited

Loading

aokolnychyi Aug 6, 2021 •

edited

Loading

aokolnychyi Aug 10, 2021 •

edited

Loading