You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/08/05 23:45:00 UTC
[GitHub] [iceberg] aokolnychyi opened a new pull request #2945: Core: Add new writer interfaces
aokolnychyi opened a new pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945
This PR adds new writer interfaces in `core` and an example of how they can be consumed in Spark 3. This will allow to write position deletes as well as write deltas in Spark. One of the major design changes is using composition over inheritance.
### 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.
### V2TaskWriter
This PR also introduces a new `TaskWriter` (I call it v2 but we better replace the existing API) and `DeltaTaskWriter` interfaces. They will be used by query engine integrations to write data from a single task. One notable difference compared to the existing code, I am using composition instead of inheritance and delegate to `TaskWriter` from query engine sinks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683860363
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ partitionComparator = Comparators.forType(partitionType);
+ completedPartitions = StructLikeSet.create(partitionType);
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711311295
##########
File path: core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java
##########
@@ -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());
Review comment:
+1 for the rename.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711326100
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -28,15 +27,17 @@
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.deletes.PositionDeleteWriter;
import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.CharSequenceWrapper;
-class SortedPosDeleteWriter<T> implements Closeable {
+class SortedPosDeleteWriter<T> implements FileWriter<PositionDelete<T>, DeleteWriteResult> {
Review comment:
I think it will. I'll update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711368342
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredFileWriter.java
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that requires the incoming records
+ * to be clustered by partition spec and partition.
+ * <p>
+ * As opposed to {@link FanoutFileWriter}, this writer keeps at most one file open to reduce
+ * the memory consumption.
+ */
+abstract class ClusteredFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private FileWriter<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711368744
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredFileWriter.java
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that requires the incoming records
+ * to be clustered by partition spec and partition.
+ * <p>
+ * As opposed to {@link FanoutFileWriter}, this writer keeps at most one file open to reduce
+ * the memory consumption.
+ */
+abstract class ClusteredFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private FileWriter<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ 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;
Review comment:
Can we move the null handling into `StructCopy` instead of doing it here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683860812
##########
File path: core/src/main/java/org/apache/iceberg/io/MixedDeltaTaskWriter.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaTaskWriter<T> extends BaseDeltaTaskWriter<T> {
Review comment:
This is what we are going to use in Spark to write deltas during MERGE INTO.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685449943
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ partitionComparator = Comparators.forType(partitionType);
+ completedPartitions = StructLikeSet.create(partitionType);
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r707767897
##########
File path: core/src/main/java/org/apache/iceberg/io/MixedDeltaWriter.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaWriter<T> extends BaseDeltaWriter<T> {
Review comment:
I agree.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683859691
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ 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;
+ currentWriter = newWriter(currentSpec, currentPartition);
+
+ } else if (partition != currentPartition && partitionComparator.compare(partition, currentPartition) != 0) {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r684536999
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredDeleteWriter.java
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public abstract class ClusteredDeleteWriter<T> extends ClusteredWriter<T, DeleteWriteResult> {
+
+ private final List<DeleteFile> deleteFiles = Lists.newArrayList();
+ private final CharSequenceSet referencedDataFiles = CharSequenceSet.empty();
+
+ @Override
+ protected void add(DeleteWriteResult result) {
+ deleteFiles.addAll(result.deleteFiles());
+ referencedDataFiles.addAll(result.referencedDataFiles());
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683858777
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
Review comment:
Instead of deriving a partition struct through `PartitionKey`, the new writer accepts `StructLike`. In some cases, this will come from a metadata column instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691609188
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
+
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ void abort() throws IOException;
+
+ Result result();
Review comment:
Removed `abort`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683858349
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
Review comment:
We support writing across multiple specs now. Background [here](https://github.com/apache/iceberg/pull/2873#discussion_r677217147).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691592846
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
Review comment:
I wanted to but we have `SortedPosDeleteWriter` that does not support giving the length. Maybe, we should just add `length` to `FileWriter` and throw an exception whenever it is not supported. Thoughts, @stevenzwu?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] pvary commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
pvary commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-904343633
> @pvary, I'll add tests and benchmarks once I split this into smaller PRs.
@aokolnychyi, sounds good, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r688275641
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
Review comment:
can we add a `long length()` API to to the `FileWriter` interface and remove this method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683857536
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
Review comment:
Naming suggestions are welcome.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-904243775
@pvary, I'll add tests and benchmarks once I split this into smaller PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685485248
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r684542082
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaTaskWriter.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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 org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaTaskWriter<T> extends V2TaskWriter<T> {
+
+ // equality delete
+ 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;
+
+ // position delete without persisting row
+ default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike partition) throws IOException {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi closed pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi closed pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685503855
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaTaskWriter.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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 org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaTaskWriter<T> extends V2TaskWriter<T> {
+
+ // equality delete
+ 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;
+
+ // position delete without persisting row
+ default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike partition) throws IOException {
Review comment:
Agreed. `UTF8String` handling is something we should probably look into independently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711403286
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<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 CDCWriter(FanoutDataWriter<T> dataWriter,
+ PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+ PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
+ Schema schema, Schema deleteSchema, Function<T, StructLike> toStructLike) {
+ 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 {
Review comment:
Could this extend `PositionDelete` so it can be passed directly into the delete writer?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711324848
##########
File path: core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java
##########
@@ -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);
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r687113315
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
+
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ void abort() throws IOException;
Review comment:
I removed `abort` and it looks like we don't really need `V2TaskWriter` anymore. So I removed that interface completely and just kept `DeltaWriter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r686299220
##########
File path: core/src/main/java/org/apache/iceberg/io/DataWriter.java
##########
@@ -57,6 +56,12 @@ public DataWriter(FileAppender<T> appender, FileFormat format, String location,
this.sortOrder = sortOrder;
}
+ @Override
+ public void write(T row) throws IOException {
Review comment:
Probably, no. I guess `IOException` was just defined in the parent interface.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691608988
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
Review comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r686301856
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
+
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ void abort() throws IOException;
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711313203
##########
File path: core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java
##########
@@ -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);
Review comment:
This class makes an assumption similar to the `ClusteredFileWriter` classes. Should we name it `OrderedPositionDeleteWriter` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711396260
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaWriter.java
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaWriter<T> extends Closeable {
+
+ // insert
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ // equality delete
+ 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;
Review comment:
Other APIs put the optional row last. What about moving spec and partition before row? Would that be too inconsistent with others?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711370088
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredDataWriter.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * A data writer capable of writing to multiple specs and partitions that requires the incoming records
+ * to be clustered by partition spec and partition.
+ */
+public class ClusteredDataWriter<T> extends ClusteredFileWriter<T, DataWriteResult> {
+
+ private final WriterFactory<T> writerFactory;
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final List<DataFile> dataFiles;
+
+ public ClusteredDataWriter(WriterFactory<T> writerFactory, OutputFileFactory fileFactory,
+ FileIO io, FileFormat fileFormat, long targetFileSizeInBytes) {
+ this.writerFactory = writerFactory;
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.dataFiles = Lists.newArrayList();
+ }
+
+ @Override
+ protected FileWriter<T, DataWriteResult> newWriter(PartitionSpec spec, StructLike partition) {
+ return new RollingDataWriter<>(writerFactory, fileFactory, io, fileFormat, targetFileSizeInBytes, spec, partition);
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711303800
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingDataWriter.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * A rolling data writer that splits incoming data into multiple files within one spec/partition.
+ */
+public class RollingDataWriter<T> extends RollingFileWriter<T, DataWriter<T>, DataWriteResult> {
+
+ private final WriterFactory<T> writerFactory;
+ private final List<DataFile> dataFiles;
+
+ public RollingDataWriter(WriterFactory<T> writerFactory, OutputFileFactory fileFactory,
+ FileIO io, FileFormat fileFormat, long targetFileSizeInBytes,
+ PartitionSpec spec, StructLike partition) {
+ super(fileFactory, io, fileFormat, targetFileSizeInBytes, spec, partition);
+ this.writerFactory = writerFactory;
+ this.dataFiles = Lists.newArrayList();
+ openCurrent();
+ }
+
+ @Override
+ protected DataWriter<T> newWriter(EncryptedOutputFile file) {
+ return writerFactory.newDataWriter(file, spec(), partition());
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685451777
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
Review comment:
Minor: I think this `add` method could be confusing since I would expect `writer.add` to write a record. How about `addResult` to be more clear?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-908682253
@openinx, congratulations!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683860734
##########
File path: core/src/main/java/org/apache/iceberg/io/MixedDeltaTaskWriter.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaTaskWriter<T> extends BaseDeltaTaskWriter<T> {
Review comment:
Naming TBD.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683860006
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ 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;
+ currentWriter = newWriter(currentSpec, currentPartition);
+
+ } else if (partition != currentPartition && partitionComparator.compare(partition, currentPartition) != 0) {
+ closeCurrent();
+ completedPartitions.add(currentPartition);
+
+ if (completedPartitions.contains(partition)) {
+ String path = spec.partitionToPath(partition);
+ throw new IllegalStateException("Already closed files for partition: " + path);
+ }
+
+ // copy the partition key as the key object is reused
+ currentPartition = partition != null ? StructCopy.copy(partition) : null;
Review comment:
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683857303
##########
File path: 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> {
Review comment:
This is essentially a copy of the existing `BaseEqualityDeltaWriter` but using new abstractions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-926981713
I am closing this one in favor of smaller PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691609431
##########
File path: 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 DeltaWriter.Result {
Review comment:
Moved.
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
+ }
+ this.currentWriter = newWriter(currentFile);
+ this.currentRows = 0;
+ }
+
+ private boolean shouldRollToNewFile() {
+ // TODO: ORC file now not support target file size before closed
+ return !fileFormat.equals(FileFormat.ORC) &&
+ currentRows % ROWS_DIVISOR == 0 && length(currentWriter) >= targetFileSizeInBytes;
+ }
+
+ private void closeCurrent() throws IOException {
+ if (currentWriter != null) {
+ currentWriter.close();
+
+ R result = currentWriter.result();
Review comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685458640
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ partitionComparator = Comparators.forType(partitionType);
+ completedPartitions = StructLikeSet.create(partitionType);
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685479827
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
Review comment:
Agreed, I'll change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691591567
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingDeleteWriter.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CharSequenceSet;
+
+abstract class RollingDeleteWriter<T, W extends FileWriter<T, DeleteWriteResult>>
Review comment:
We currently have 2 delete writers: `EqualityDeleteWriter` and `PositionDeleteWriter`. I made them both extend the newly added `FileWriter` interface.
Whenever I needed an arbitrary delete writer, I used `FileWriter<T, DeleteWriteResult>`. We could add a separate interface called `DeleteWriter` but it will be empty and just parameterize `FileWriter`.
I don't have a preference here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683862056
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
##########
@@ -531,68 +536,111 @@ protected WriterFactory(Broadcast<Table> tableBroadcast, FileFormat format, long
@Override
public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
Table table = tableBroadcast.value();
-
- OutputFileFactory fileFactory = OutputFileFactory.builderFor(table, partitionId, taskId).format(format).build();
- SparkAppenderFactory appenderFactory = SparkAppenderFactory.builderFor(table, writeSchema, dsSchema).build();
-
PartitionSpec spec = table.spec();
FileIO io = table.io();
+ OutputFileFactory fileFactory = OutputFileFactory.builderFor(table, partitionId, taskId)
Review comment:
This is an example of how to use the new writers in Spark 3.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-904794157
@rdblue @openinx, do you are agree with the direction?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r688272071
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingDeleteWriter.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CharSequenceSet;
+
+abstract class RollingDeleteWriter<T, W extends FileWriter<T, DeleteWriteResult>>
Review comment:
I comparing this to the `RollingDataWriter`, should we introduce a `DeleteWriter` (similar to `DataWriter`)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r695374734
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
Review comment:
yeah. I am in favor of having `SortedPosDeleteWriter` throwing not supported exception
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711294318
##########
File path: core/src/main/java/org/apache/iceberg/io/FileWriter.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+
+/**
+ * A writer capable of writing files of a single type (i.e. data/delete) to one spec/partition.
+ * <p>
+ * As opposed to {@link FileAppender}, this interface should be implemented by classes that not only
+ * append records to files but actually produce {@link DataFile}s or {@link DeleteFile}s objects
+ * with Iceberg metadata. Implementations may wrap {@link FileAppender}s with extra information
+ * such as spec, partition, sort order ID needed to construct {@link DataFile}s or {@link DeleteFile}s.
+ */
+public interface FileWriter<T, R> extends Closeable {
Review comment:
Can you add docs for the T and R types?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683857820
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * A writer capable of writing to multiple specs and partitions ensuring the incoming records are properly clustered.
Review comment:
This one is similar to our existing `PartitionedWriter` but with some notable differences I'll mention below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r684556830
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
Review comment:
Maybe, we better call it something else and keep the old hierarchy for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683861409
##########
File path: core/src/main/java/org/apache/iceberg/io/PartitionAwareWriter.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+/**
+ * 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 {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683858777
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private Writer<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
Review comment:
Instead of deriving `PartitionKey`, the new writer accepts `StructLike`. In some cases, this will come from a metadata column instead of being derived through `PartitionKey`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r686317001
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
+
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ void abort() throws IOException;
+
+ Result result();
Review comment:
One benefit of removing `abort` from here is that we won't have to reason about it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685449277
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
Review comment:
I like "clustered" since that's the assumption.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685502668
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
+
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ void abort() throws IOException;
Review comment:
What is the expectation for calling abort? Should the writer already be closed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-908428481
Thanks for pinging me @aokolnychyi , I've been on paternity leave for a while. So seems I've blocked this PR, I will take a look at this PR tomorrow or the day after tomorrow. (Sorry for the delay).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685506430
##########
File path: core/src/main/java/org/apache/iceberg/io/Writer.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+
+/**
+ * A writer capable of writing files of a single type (i.e. data/delete) within one spec/partition.
+ */
+public interface Writer<T, R> extends Closeable {
+
+ default void write(Iterable<T> rows) throws IOException {
+ for (T row : rows) {
+ write(row);
+ }
+ }
+
+ void write(T row) throws IOException;
+
+ R result();
Review comment:
Does `result` need to be called after `close`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691586199
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter;
+ private final StructProjection keyProjection;
+ private final Map<StructLike, PartitionAwarePathOffset> insertedRows;
Review comment:
One distinction compared to the existing CDC writer is that we track inserted rows across partitions. For example, if we add a record with `ID=1` into partition `year=2020` and then add another record with `ID=1`, we will write a position delete for `ID=1` and `year=2020`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r687115258
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaWriter.java
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaWriter<T> extends Closeable {
+
+ // insert
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ // equality delete
+ 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;
+
+ // position delete without persisting row
+ default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike partition) throws IOException {
+ delete(path, pos, null, spec, partition);
+ }
+
+ Result result();
+
+ interface Result extends Serializable {
+ List<DataFile> dataFiles();
Review comment:
I debated a little bit whether this should be lists or arrays. I went with lists (non-guava) as we aggregate data files into lists and converting that into arrays would require a full copy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r700196710
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingEqualityDeleteWriter.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+
+/**
+ * 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>> {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711404297
##########
File path: core/src/main/java/org/apache/iceberg/io/BaseDeltaWriter.java
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.Lists;
+import org.apache.iceberg.util.CharSequenceSet;
+
+abstract class BaseDeltaWriter<T> implements DeltaWriter<T> {
+
+ private final List<DataFile> dataFiles = Lists.newArrayList();
+ private final List<DeleteFile> deleteFiles = Lists.newArrayList();
+ private final CharSequenceSet referencedDataFiles = CharSequenceSet.empty();
+
+ private boolean closed = false;
+
+ @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;
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711304291
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
Review comment:
This should call `newOutputFile(spec, partition)`, not this one that uses the default spec.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711397818
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
Review comment:
This also needs some Javadoc to explain the context.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711395322
##########
File path: core/src/main/java/org/apache/iceberg/io/FanoutFileWriter.java
##########
@@ -0,0 +1,107 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.StructLikeMap;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that keeps files for each
+ * seen spec/partition pair open until this writer is closed.
+ * <p>
+ * As opposed to {@link ClusteredFileWriter}, this writer does not requite the incoming records
+ * to be clustered by partition spec and partition. The downside is potentially larger memory
+ * consumption.
+ */
+abstract class FanoutFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Map<Integer, Map<StructLike, FileWriter<T, R>>> writers = Maps.newHashMap();
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ public CharSequence currentPath(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentPath();
+ }
+
+ public long currentPosition(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentRows();
+ }
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ FileWriter<T, R> writer = writer(spec, partition);
+ writer.write(row);
+ }
+
+ private FileWriter<T, R> writer(PartitionSpec spec, StructLike partition) {
+ 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) {
Review comment:
Nevermind, it's the copy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r707766786
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaWriter.java
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaWriter<T> extends Closeable {
+
+ // insert
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ // equality delete
+ 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;
+
+ // position delete without persisting row
+ default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike partition) throws IOException {
+ delete(path, pos, null, spec, partition);
+ }
+
+ Result result();
+
+ interface Result extends Serializable {
+ List<DataFile> dataFiles();
Review comment:
Correct, I remember that issue and that's why I used non-guava mutable lists. I agree it would be safer to explicitly use arrays but I am worried about the cost of creating arrays from lists. We have to use lists to accumulate files as they are written.
Am I overthinking it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r707767440
##########
File path: core/src/main/java/org/apache/iceberg/io/FanoutFileWriter.java
##########
@@ -0,0 +1,107 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.StructLikeMap;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that keeps files for each
+ * seen spec/partition pair open until this writer is closed.
+ * <p>
+ * As opposed to {@link ClusteredFileWriter}, this writer does not requite the incoming records
+ * to be clustered by partition spec and partition. The downside is potentially larger memory
+ * consumption.
+ */
+abstract class FanoutFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Map<Integer, Map<StructLike, FileWriter<T, R>>> writers = Maps.newHashMap();
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ public CharSequence currentPath(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentPath();
+ }
+
+ public long currentPosition(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentRows();
+ }
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ FileWriter<T, R> writer = writer(spec, partition);
+ writer.write(row);
+ }
+
+ private FileWriter<T, R> writer(PartitionSpec spec, StructLike partition) {
+ 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;
+ }
Review comment:
I can change that. No preference from my side here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r684536999
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredDeleteWriter.java
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public abstract class ClusteredDeleteWriter<T> extends ClusteredWriter<T, DeleteWriteResult> {
+
+ private final List<DeleteFile> deleteFiles = Lists.newArrayList();
+ private final CharSequenceSet referencedDataFiles = CharSequenceSet.empty();
+
+ @Override
+ protected void add(DeleteWriteResult result) {
+ deleteFiles.addAll(result.deleteFiles());
+ referencedDataFiles.addAll(result.referencedDataFiles());
Review comment:
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 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-908682253
@openinx, congratulations!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711397641
##########
File path: core/src/main/java/org/apache/iceberg/io/MixedDeltaWriter.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaWriter<T> extends BaseDeltaWriter<T> {
Review comment:
I think this needs Javadoc and a better name that shows this only accepts position deletes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711402164
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<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 CDCWriter(FanoutDataWriter<T> dataWriter,
+ PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+ PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
Review comment:
I think that this needs to be `FanoutSortedPositionDeleteWriter` because the position deletes could be in any order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r707758906
##########
File path: core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java
##########
@@ -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());
Review comment:
I agree. I'll do that rename when I split this into smaller PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691583535
##########
File path: 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 DeltaWriter.Result {
Review comment:
I'll move it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r686300400
##########
File path: core/src/main/java/org/apache/iceberg/io/Writer.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+
+/**
+ * A writer capable of writing files of a single type (i.e. data/delete) within one spec/partition.
+ */
+public interface Writer<T, R> extends Closeable {
+
+ default void write(Iterable<T> rows) throws IOException {
+ for (T row : rows) {
+ write(row);
+ }
+ }
+
+ void write(T row) throws IOException;
+
+ R result();
Review comment:
Yes, that's ensured in all writer implementations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] pvary commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
pvary commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-903932242
> ```
> Benchmark Mode Cnt Score Error Units
> TaskWriterParquetBenchmark.writePartitionedDataNewFanoutWriter ss 5 10.432 ± 0.382 s/op
> ```
Would it worth to add this benchmark to the codebase?
I need to check again, but I might have to change the Spark Orc writer-s, and I would like to make sure that the performance remains the same, so having a good benchmark/ baseline would be good.
Thanks,
Peter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-894524869
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
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r696232779
##########
File path: core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java
##########
@@ -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
Review comment:
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.
##########
File path: core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java
##########
@@ -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());
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r686301856
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
+
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ void abort() throws IOException;
Review comment:
This essentially mimics the current interface we have. I am happy to reconsider having `abort` in our writers. I guess the idea is to reuse the cleanup logic in all query engine integrations. We could have some utility methods instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-908613323
No worries, @openinx! Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711307328
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
+ }
+ this.currentWriter = newWriter(currentFile);
+ this.currentRows = 0;
+ }
+
+ private boolean shouldRollToNewFile() {
+ // TODO: ORC file now not support target file size before closed
+ return !fileFormat.equals(FileFormat.ORC) &&
+ currentRows % ROWS_DIVISOR == 0 && length(currentWriter) >= targetFileSizeInBytes;
+ }
+
+ private void closeCurrent() throws IOException {
+ if (currentWriter != null) {
+ currentWriter.close();
+
+ if (currentRows == 0L) {
+ io.deleteFile(currentFile.encryptingOutputFile());
+ } else {
+ R result = currentWriter.result();
+ addResult(result);
Review comment:
Minor: do we need `result` or can we just call `addResult(currentWriter.result())`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r707769076
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingEqualityDeleteWriter.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+
+/**
+ * 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>> {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711302925
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingDataWriter.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * A rolling data writer that splits incoming data into multiple files within one spec/partition.
+ */
+public class RollingDataWriter<T> extends RollingFileWriter<T, DataWriter<T>, DataWriteResult> {
+
+ private final WriterFactory<T> writerFactory;
+ private final List<DataFile> dataFiles;
+
+ public RollingDataWriter(WriterFactory<T> writerFactory, OutputFileFactory fileFactory,
+ FileIO io, FileFormat fileFormat, long targetFileSizeInBytes,
+ PartitionSpec spec, StructLike partition) {
+ super(fileFactory, io, fileFormat, targetFileSizeInBytes, spec, partition);
+ this.writerFactory = writerFactory;
+ this.dataFiles = Lists.newArrayList();
+ openCurrent();
+ }
+
+ @Override
+ protected DataWriter<T> newWriter(EncryptedOutputFile file) {
+ return writerFactory.newDataWriter(file, spec(), partition());
Review comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711322322
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
+ }
+ this.currentWriter = newWriter(currentFile);
+ this.currentRows = 0;
+ }
+
+ private boolean shouldRollToNewFile() {
+ // TODO: ORC file now not support target file size before closed
+ return !fileFormat.equals(FileFormat.ORC) &&
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r686315345
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * 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> {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685506040
##########
File path: core/src/main/java/org/apache/iceberg/io/DataWriter.java
##########
@@ -57,6 +56,12 @@ public DataWriter(FileAppender<T> appender, FileFormat format, String location,
this.sortOrder = sortOrder;
}
+ @Override
+ public void write(T row) throws IOException {
Review comment:
Do we need to throw `IOException` if the current writer does not?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711400309
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<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 CDCWriter(FanoutDataWriter<T> dataWriter,
+ PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+ PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
+ Schema schema, Schema deleteSchema, Function<T, StructLike> toStructLike) {
+ this.dataWriter = dataWriter;
+ this.equalityDeleteWriter = equalityDeleteWriter;
+ this.positionDeleteWriter = positionDeleteWriter;
+ this.positionDelete = new PositionDelete<>();
Review comment:
Nit: `PositionDelete.create()`
The constructor should probably be made private.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711304656
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
Review comment:
In fact, this version should probably be deprecated so that we can remove it once we move over to these writers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711372916
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredFileWriter.java
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that requires the incoming records
+ * to be clustered by partition spec and partition.
+ * <p>
+ * As opposed to {@link FanoutFileWriter}, this writer keeps at most one file open to reduce
+ * the memory consumption.
+ */
+abstract class ClusteredFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private FileWriter<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
Review comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711320125
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingDataWriter.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * A rolling data writer that splits incoming data into multiple files within one spec/partition.
+ */
+public class RollingDataWriter<T> extends RollingFileWriter<T, DataWriter<T>, DataWriteResult> {
+
+ private final WriterFactory<T> writerFactory;
+ private final List<DataFile> dataFiles;
+
+ public RollingDataWriter(WriterFactory<T> writerFactory, OutputFileFactory fileFactory,
+ FileIO io, FileFormat fileFormat, long targetFileSizeInBytes,
+ PartitionSpec spec, StructLike partition) {
+ super(fileFactory, io, fileFormat, targetFileSizeInBytes, spec, partition);
+ this.writerFactory = writerFactory;
+ this.dataFiles = Lists.newArrayList();
+ openCurrent();
+ }
+
+ @Override
+ protected DataWriter<T> newWriter(EncryptedOutputFile file) {
+ return writerFactory.newDataWriter(file, spec(), partition());
Review comment:
I don't have a strong opinion here. If you feel it is not worth it, I can surely adapt.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711314663
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -28,15 +27,17 @@
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.deletes.PositionDeleteWriter;
import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.CharSequenceWrapper;
-class SortedPosDeleteWriter<T> implements Closeable {
+class SortedPosDeleteWriter<T> implements FileWriter<PositionDelete<T>, DeleteWriteResult> {
Review comment:
Should this wrap an `OrderedPositionDeleteWriter` instead of managing its own appender? Would that help the issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r707757939
##########
File path: core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java
##########
@@ -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
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r687116592
##########
File path: core/src/main/java/org/apache/iceberg/io/FileWriter.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+
+/**
+ * A writer capable of writing files of a single type (i.e. data/delete) to one spec/partition.
+ * <p>
+ * As opposed to {@link FileAppender}, this interface should be implemented by classes that not only
+ * append records to files but actually produce {@link DataFile}s or {@link DeleteFile}s objects
+ * with Iceberg metadata. Implementations may wrap {@link FileAppender}s with extra information
+ * such as spec, partition, sort order ID needed to construct {@link DataFile}s or {@link DeleteFile}s.
+ */
+public interface FileWriter<T, R> extends Closeable {
Review comment:
@rdblue, I renamed this interface as `FileWriter`. My main motivation for that was to be more specific and follow the naming we use in `FileAppender`. I also added docs to describe the difference between the two.
It may mean we should rename our recently added `WriterFactory` into `FileWriterFactory`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r684542082
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaTaskWriter.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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 org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaTaskWriter<T> extends V2TaskWriter<T> {
+
+ // equality delete
+ 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;
+
+ // position delete without persisting row
+ default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike partition) throws IOException {
Review comment:
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.
Thoughts, @rdblue @RussellSpitzer?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r688273307
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
+ }
+ this.currentWriter = newWriter(currentFile);
+ this.currentRows = 0;
+ }
+
+ private boolean shouldRollToNewFile() {
+ // TODO: ORC file now not support target file size before closed
+ return !fileFormat.equals(FileFormat.ORC) &&
+ currentRows % ROWS_DIVISOR == 0 && length(currentWriter) >= targetFileSizeInBytes;
+ }
+
+ private void closeCurrent() throws IOException {
+ if (currentWriter != null) {
+ currentWriter.close();
+
+ R result = currentWriter.result();
Review comment:
nit: move this inside the else
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-896326936
@openinx, could you please take a look whenever you get a second?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683861628
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * 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> {
Review comment:
Similar to our existing `BaseRollingWriter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r687119370
##########
File path: core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java
##########
@@ -28,15 +27,17 @@
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.deletes.PositionDeleteWriter;
import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.util.CharSequenceSet;
import org.apache.iceberg.util.CharSequenceWrapper;
-class SortedPosDeleteWriter<T> implements Closeable {
+class SortedPosDeleteWriter<T> implements FileWriter<PositionDelete<T>, DeleteWriteResult> {
Review comment:
@openinx @rdblue, this writer is slightly different from other delete writers we have. Currently, we cannot construct it using `WriterFactory`. I did not add it because it is not possible to build this writer via `Parquet`, `Avro` classes. I'd like to know your thoughts whether we should extend `Parquet`, `Avro`, `WriterFactory` classes to support this one or we consider its purpose fairly narrow so that we will still build `FileAppender` manually and not through `WriterFactory`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683862318
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
Review comment:
We have `V2` in the name temporarily. We already have `TaskWriter` and I did not want to change that in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691586502
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter;
+ private final StructProjection keyProjection;
+ private final Map<StructLike, PartitionAwarePathOffset> insertedRows;
Review comment:
That's why the CDC writer is aware of partitions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691593331
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
+ }
+ this.currentWriter = newWriter(currentFile);
+ this.currentRows = 0;
+ }
+
+ private boolean shouldRollToNewFile() {
+ // TODO: ORC file now not support target file size before closed
+ return !fileFormat.equals(FileFormat.ORC) &&
+ currentRows % ROWS_DIVISOR == 0 && length(currentWriter) >= targetFileSizeInBytes;
+ }
+
+ private void closeCurrent() throws IOException {
+ if (currentWriter != null) {
+ currentWriter.close();
+
+ R result = currentWriter.result();
Review comment:
Will do.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-901437006
I pushed a few updates. I used the code from this PR to implement position-based merge-on-read DELETE for Spark and the implementation managed to pass all tests. I am ready to split this into smaller PRs with tests and benchmarks but we will need @openinx's approval to proceed.
I happy to jump on a short call if needed too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r707764349
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<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 CDCWriter(FanoutDataWriter<T> dataWriter,
+ PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+ PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
+ Schema schema, Schema deleteSchema, Function<T, StructLike> toStructLike) {
+ 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);
Review comment:
Since this is something that we will have to do for all insert rows, I was trying to avoid generating an extra object to reduce the memory consumption.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711369229
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredFileWriter.java
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that requires the incoming records
+ * to be clustered by partition spec and partition.
+ * <p>
+ * As opposed to {@link FanoutFileWriter}, this writer keeps at most one file open to reduce
+ * the memory consumption.
+ */
+abstract class ClusteredFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private FileWriter<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ 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;
+ currentWriter = newWriter(currentSpec, currentPartition);
+
+ } else if (partition != currentPartition && partitionComparator.compare(partition, currentPartition) != 0) {
+ closeCurrent();
+ completedPartitions.add(currentPartition);
+
+ if (completedPartitions.contains(partition)) {
+ String path = spec.partitionToPath(partition);
+ throw new IllegalStateException("Already closed files for partition: " + path);
+ }
+
+ // copy the partition key as the key object is reused
+ currentPartition = partition != null ? StructCopy.copy(partition) : null;
+ currentWriter = newWriter(currentSpec, currentPartition);
+ }
+
+ currentWriter.write(row);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!closed) {
+ closeCurrent();
+ this.closed = true;
+ }
+ }
+
+ private void closeCurrent() throws IOException {
+ if (currentWriter != null) {
+ currentWriter.close();
+
+ R result = currentWriter.result();
+ addResult(result);
Review comment:
Useless variable, `result`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-908613323
No worries, @openinx! Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711318475
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingDataWriter.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * A rolling data writer that splits incoming data into multiple files within one spec/partition.
+ */
+public class RollingDataWriter<T> extends RollingFileWriter<T, DataWriter<T>, DataWriteResult> {
+
+ private final WriterFactory<T> writerFactory;
+ private final List<DataFile> dataFiles;
+
+ public RollingDataWriter(WriterFactory<T> writerFactory, OutputFileFactory fileFactory,
+ FileIO io, FileFormat fileFormat, long targetFileSizeInBytes,
+ PartitionSpec spec, StructLike partition) {
+ super(fileFactory, io, fileFormat, targetFileSizeInBytes, spec, partition);
+ this.writerFactory = writerFactory;
+ this.dataFiles = Lists.newArrayList();
+ openCurrent();
+ }
+
+ @Override
+ protected DataWriter<T> newWriter(EncryptedOutputFile file) {
+ return writerFactory.newDataWriter(file, spec(), partition());
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711372649
##########
File path: core/src/main/java/org/apache/iceberg/io/FanoutFileWriter.java
##########
@@ -0,0 +1,107 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.StructLikeMap;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that keeps files for each
+ * seen spec/partition pair open until this writer is closed.
+ * <p>
+ * As opposed to {@link ClusteredFileWriter}, this writer does not requite the incoming records
+ * to be clustered by partition spec and partition. The downside is potentially larger memory
+ * consumption.
+ */
+abstract class FanoutFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Map<Integer, Map<StructLike, FileWriter<T, R>>> writers = Maps.newHashMap();
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ public CharSequence currentPath(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentPath();
+ }
+
+ public long currentPosition(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentRows();
+ }
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ FileWriter<T, R> writer = writer(spec, partition);
+ writer.write(row);
+ }
+
+ private FileWriter<T, R> writer(PartitionSpec spec, StructLike partition) {
+ 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) {
Review comment:
Can we not use `computeIfAbsent` for the `StructLikeMap` as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-908428481
Thanks for pinging me @aokolnychyi , I've been on paternity leave for a while. So seems I've blocked this PR, I will take a look at this PR tomorrow or the day after tomorrow. (Sorry for the delay).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711320691
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
Review comment:
Good catch, I think I did this PR before we had the new method. I'll update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711290069
##########
File path: core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java
##########
@@ -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
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711367239
##########
File path: core/src/main/java/org/apache/iceberg/io/ClusteredFileWriter.java
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that requires the incoming records
+ * to be clustered by partition spec and partition.
+ * <p>
+ * As opposed to {@link FanoutFileWriter}, this writer keeps at most one file open to reduce
+ * the memory consumption.
+ */
+abstract class ClusteredFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Set<Integer> completedSpecs = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private FileWriter<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrent();
+ completedSpecs.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecs.contains(spec.specId())) {
+ throw new IllegalStateException("Already closed files for spec: " + spec.specId());
+ }
+
+ Types.StructType partitionType = spec.partitionType();
+
+ currentSpec = spec;
+ partitionComparator = Comparators.forType(partitionType);
+ completedPartitions = StructLikeSet.create(partitionType);
Review comment:
This seems to assume that partitions will be grouped by spec. Do you think that's likely?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r700177068
##########
File path: core/src/main/java/org/apache/iceberg/io/FanoutFileWriter.java
##########
@@ -0,0 +1,107 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.StructLikeMap;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that keeps files for each
+ * seen spec/partition pair open until this writer is closed.
+ * <p>
+ * As opposed to {@link ClusteredFileWriter}, this writer does not requite the incoming records
+ * to be clustered by partition spec and partition. The downside is potentially larger memory
+ * consumption.
+ */
+abstract class FanoutFileWriter<T, R> implements PartitionAwareFileWriter<T, R> {
+
+ private final Map<Integer, Map<StructLike, FileWriter<T, R>>> writers = Maps.newHashMap();
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ public CharSequence currentPath(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentPath();
+ }
+
+ public long currentPosition(PartitionSpec spec, StructLike partition) {
+ return ((RollingFileWriter<?, ?, ?>) writer(spec, partition)).currentRows();
+ }
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ FileWriter<T, R> writer = writer(spec, partition);
+ writer.write(row);
+ }
+
+ private FileWriter<T, R> writer(PartitionSpec spec, StructLike partition) {
+ 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;
+ }
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r688247826
##########
File path: 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 DeltaWriter.Result {
Review comment:
should this be a static nested class as `BaseDeltaWriter.Result` to match the style of `DeltaWriter.Result`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683862318
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
Review comment:
`V2` in the name is temporarily. We already have `TaskWriter` and I did not want to change that in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r696107895
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
Review comment:
Sounds good to me. I'll change it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r700155639
##########
File path: core/src/main/java/org/apache/iceberg/io/MixedDeltaWriter.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.PositionDelete;
+
+public class MixedDeltaWriter<T> extends BaseDeltaWriter<T> {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r700148679
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaWriter.java
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaWriter<T> extends Closeable {
+
+ // insert
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ // equality delete
+ 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;
+
+ // position delete without persisting row
+ default void delete(CharSequence path, long pos, PartitionSpec spec, StructLike partition) throws IOException {
+ delete(path, pos, null, spec, partition);
+ }
+
+ Result result();
+
+ interface Result extends Serializable {
+ List<DataFile> dataFiles();
Review comment:
The `Result` will be serialized and transferred in the network to spark driver or flink IcebergFilesCommitter, we always choose to use arrays before because we wanna to get the rid of the kryo serialize issue about immutable array lists. See the similar comment here: https://github.com/apache/iceberg/pull/1145#discussion_r448537766
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r683860643
##########
File path: core/src/main/java/org/apache/iceberg/io/FanoutWriter.java
##########
@@ -0,0 +1,105 @@
+/*
+ * 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 org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.StructLikeMap;
+
+/**
+ * A writer capable of writing to multiple specs and partitions that keeps writers for each
+ * seen spec/partition pair open until this writer is closed.
+ */
+public abstract class FanoutWriter<T, R> implements PartitionAwareWriter<T, R> {
+
+ private final Map<Integer, Map<StructLike, Writer<T, R>>> writers = Maps.newHashMap();
+ private boolean closed = false;
+
+ protected abstract Writer<T, R> newWriter(PartitionSpec spec, StructLike partition);
+
+ protected abstract void add(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public CharSequence currentPath(PartitionSpec spec, StructLike partition) {
+ return ((RollingWriter<?, ?, ?>) writer(spec, partition)).currentPath();
+ }
+
+ @Override
+ public long currentPosition(PartitionSpec spec, StructLike partition) {
+ return ((RollingWriter<?, ?, ?>) writer(spec, partition)).currentRows();
+ }
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) throws IOException {
+ Writer<T, R> writer = writer(spec, partition);
+ writer.write(row);
+ }
+
+ private Writer<T, R> writer(PartitionSpec spec, StructLike partition) {
+ Map<StructLike, Writer<T, R>> specWriters = writers.computeIfAbsent(
+ spec.specId(),
+ id -> StructLikeMap.create(spec.partitionType()));
Review comment:
Using `StructLikeMap` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#issuecomment-893899826
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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r685502969
##########
File path: core/src/main/java/org/apache/iceberg/io/V2TaskWriter.java
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+public interface V2TaskWriter<T> extends Closeable {
+
+ void insert(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ void abort() throws IOException;
+
+ Result result();
Review comment:
What should result return if the writer is aborted?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711402921
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<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 CDCWriter(FanoutDataWriter<T> dataWriter,
+ PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+ PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
+ Schema schema, Schema deleteSchema, Function<T, StructLike> toStructLike) {
+ 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() {
Review comment:
This is called `position` nearly everywhere else. Why call it `rowOffset` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711403612
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<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 CDCWriter(FanoutDataWriter<T> dataWriter,
+ PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+ PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
+ Schema schema, Schema deleteSchema, Function<T, StructLike> toStructLike) {
+ 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 {
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711306453
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
+ }
+ this.currentWriter = newWriter(currentFile);
+ this.currentRows = 0;
+ }
+
+ private boolean shouldRollToNewFile() {
+ // TODO: ORC file now not support target file size before closed
+ return !fileFormat.equals(FileFormat.ORC) &&
Review comment:
(Not urgent to fix in this PR)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r711305832
##########
File path: core/src/main/java/org/apache/iceberg/io/RollingFileWriter.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * A rolling writer capable of splitting incoming data/deletes into multiple files within one spec/partition.
+ */
+abstract class RollingFileWriter<T, W extends FileWriter<T, R>, R> implements FileWriter<T, R> {
+ private static final int ROWS_DIVISOR = 1000;
+
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final FileFormat fileFormat;
+ private final long targetFileSizeInBytes;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+
+ private EncryptedOutputFile currentFile = null;
+ private W currentWriter = null;
+ private long currentRows = 0;
+
+ private boolean closed = false;
+
+ protected RollingFileWriter(OutputFileFactory fileFactory, FileIO io, FileFormat fileFormat,
+ long targetFileSizeInBytes, PartitionSpec spec, StructLike partition) {
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.fileFormat = fileFormat;
+ this.targetFileSizeInBytes = targetFileSizeInBytes;
+ this.spec = spec;
+ this.partition = partition;
+ }
+
+ protected abstract W newWriter(EncryptedOutputFile file);
+
+ protected abstract long length(W writer);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ protected PartitionSpec spec() {
+ return spec;
+ }
+
+ protected StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public void write(T row) throws IOException {
+ currentWriter.write(row);
+ currentRows++;
+
+ if (shouldRollToNewFile()) {
+ closeCurrent();
+ openCurrent();
+ }
+ }
+
+ public CharSequence currentPath() {
+ Preconditions.checkNotNull(currentFile, "The currentFile shouldn't be null");
+ return currentFile.encryptingOutputFile().location();
+ }
+
+ public long currentRows() {
+ return currentRows;
+ }
+
+ protected void openCurrent() {
+ if (partition == null) {
+ this.currentFile = fileFactory.newOutputFile();
+ } else {
+ this.currentFile = fileFactory.newOutputFile(partition);
+ }
+ this.currentWriter = newWriter(currentFile);
+ this.currentRows = 0;
+ }
+
+ private boolean shouldRollToNewFile() {
+ // TODO: ORC file now not support target file size before closed
+ return !fileFormat.equals(FileFormat.ORC) &&
Review comment:
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?
```java
if (fileFormat == FileFormat.ORC) {
return new DataWriter(...);
} else {
return new RollingDataWriter(...);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r695366360
##########
File path: core/src/main/java/org/apache/iceberg/io/CDCWriter.java
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 CDCWriter<T> extends BaseDeltaWriter<T> {
+
+ private final FanoutDataWriter<T> dataWriter;
+ private final PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter;
+ private final PartitionAwareFileWriter<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 CDCWriter(FanoutDataWriter<T> dataWriter,
+ PartitionAwareFileWriter<T, DeleteWriteResult> equalityDeleteWriter,
+ PartitionAwareFileWriter<PositionDelete<T>, DeleteWriteResult> positionDeleteWriter,
+ Schema schema, Schema deleteSchema, Function<T, StructLike> toStructLike) {
+ 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);
Review comment:
should we merge the two calls into one method?
```
FilePathPosition pathPosition = dataWriter.filePathPosition(spec, partition);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] stevenzwu commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r688264241
##########
File path: core/src/main/java/org/apache/iceberg/io/PartitionAwareFileWriter.java
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+/**
+ * A writer capable of writing files of a single type (i.e. data/delete) to multiple specs and partitions.
+ * <p>
+ * As opposed to {@link FileWriter}, this interface should be implemented by writers that are not
+ * limited to writing to a single spec/partition. Implementations may under the hood use other
+ * {@link FileWriter}s for writing to a single spec/partition.
+ */
+public interface PartitionAwareFileWriter<T, R> extends Closeable {
+
+ /**
+ * Writes a row to the provided spec/partition.
+ *
+ * @param row a data or delete record
+ * @param spec a partition spec
+ * @param partition a partition or null if the spec is unpartitioned
+ * @throws IOException in case of an error during the write process
+ */
+ void write(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ /**
+ * Returns a result that contains information about written {@link DataFile}s or {@link DeleteFile}s.
+ * The result is valid only after the writer is closed.
Review comment:
I know the benefit of standard Closeable and AutoCloseable. On the other hand, this comment also suggests that we should combine the close and result as a single method like
```
R close() throws IOException;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r687114449
##########
File path: core/src/main/java/org/apache/iceberg/io/DeltaWriter.java
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.CharSequenceSet;
+
+public interface DeltaWriter<T> extends Closeable {
Review comment:
@rdblue, that's what I plan to delegate to from Spark's `DeltaWriter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2945: Core: Add new writer interfaces
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2945:
URL: https://github.com/apache/iceberg/pull/2945#discussion_r691588592
##########
File path: core/src/main/java/org/apache/iceberg/io/PartitionAwareFileWriter.java
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+
+/**
+ * A writer capable of writing files of a single type (i.e. data/delete) to multiple specs and partitions.
+ * <p>
+ * As opposed to {@link FileWriter}, this interface should be implemented by writers that are not
+ * limited to writing to a single spec/partition. Implementations may under the hood use other
+ * {@link FileWriter}s for writing to a single spec/partition.
+ */
+public interface PartitionAwareFileWriter<T, R> extends Closeable {
+
+ /**
+ * Writes a row to the provided spec/partition.
+ *
+ * @param row a data or delete record
+ * @param spec a partition spec
+ * @param partition a partition or null if the spec is unpartitioned
+ * @throws IOException in case of an error during the write process
+ */
+ void write(T row, PartitionSpec spec, StructLike partition) throws IOException;
+
+ /**
+ * Returns a result that contains information about written {@link DataFile}s or {@link DeleteFile}s.
+ * The result is valid only after the writer is closed.
Review comment:
I mostly followed the pattern we use in other places such as `DataWriter` and `EqualityDeleteWriter`. I guess we have a very similar design for our appenders too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org