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/10/20 03:45:42 UTC

[GitHub] [iceberg] Reo-LEI opened a new pull request #3323: Flink: Incrementally rewrite data files in streaming.

Reo-LEI opened a new pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323


   ### Description
   This PR is base on the approach of @rdblue [comment](https://github.com/apache/iceberg/pull/2867#issuecomment-891978223) and trying to incrementally rewrite committed data files.
   
   ### Implementation
   In this PR, the parallel `IcebergStreamRewriter` and single-parallelism `IcebergRewriteFilesCommitter` will append to `IcebergFilesCommitter`. 
   ```
                                 +-> IcebergStreamRewriter --+ 
   IcebergFilesCommitter --Hash--+-> IcebergStreamRewriter --+--Rebalance--> IcebergRewriteFilesCommitter
                                 +-> IcebergStreamRewriter --+ 
   ```
   - At the beginning, the committed data files and related delete files will be group by partition as `CommitResult` and And distribute to the `IcebergStreamRewriter ` according to the partition. 
   - `IcebergStreamRewriter` will collect the `CommitResult`, write a tmp `DeltaManifests` to reference all committed data/delete files and append this tmp manifest to a `DataFileGroup`. Once a file group reache a rewrite condition(file num / file size), the rewrite will be executed and emit a `RewriteResult` to `IcebergRewriteFilesCommitter`.
   - `IcebergRewriteFilesCommitter` will collect all `RewriteResult` and stream/batch commit it in serial.  
   
   
   ### Note
   - File group of a partition rewrite failed will not drop the file group or stop the flink job. `IcebergStreamRewriter` will retry rewrite this file group when this group have new data file append.
   -  `IcebergRewriteFilesCommitter` will trying commit all ckpt-1 rewrite result before ckpt(commit all remaining rewrite result in `prepareSnapshotPreBarrier`), to prevent the rewritten file will be modified at the ckpt but rewrite result not commit yet.
   
   ### Configuration
   - `flink.rewrite.max-files-count` 
   - `flink.rewrite.target-file-size-bytes`
   - `flink.rewrite.commit-groups-size` 


-- 
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] jackye1995 commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961692478


   had a lot of thinking around this PR, I decide to write down in a doc, @openinx @rdblue @stevenzwu please let me know what you think!
   
   https://docs.google.com/document/d/18N-xwZasXLNEl2407xT1oD08Mv9Kd3p0xMX7ZJFyV20/edit?usp=sharing


-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735329129



##########
File path: api/src/main/java/org/apache/iceberg/types/TypeUtil.java
##########
@@ -282,10 +282,10 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
 
     switch (from.typeId()) {
       case INTEGER:
-        return to == Types.LongType.get();
+        return to.equals(Types.LongType.get());

Review comment:
       The root cause is when comparing two [PartitionData](https://github.com/apache/iceberg/blob/24896d84c586c6339752a69773409219c0c0ac02/core/src/main/java/org/apache/iceberg/PartitionData.java#L188) for equality, we need to compare all type of partition. Currently, we compare whether the two `PrimitiveType` are equal, we compare the singleton PrimitiveType instance, once this object is serialized and deserialized, the new instance will no longer equal to the old instance, event they are the same type.
   
   For fix this probelm, I add [equal](https://github.com/apache/iceberg/blob/6439bbb2fe8ff84d6bcf2d26f764a61b6e04cf8c/api/src/main/java/org/apache/iceberg/types/Type.java#L117) and [hashcode](https://github.com/apache/iceberg/blob/6439bbb2fe8ff84d6bcf2d26f764a61b6e04cf8c/api/src/main/java/org/apache/iceberg/types/Type.java#L128) method for `PrimitiveType`. And I found that will break some [style checks](https://github.com/apache/iceberg/runs/3953641733?check_suite_focus=true#step:4:313), so I need to chang this.




-- 
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] Reo-LEI edited a comment on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI edited a comment on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962011180

I‘m very grateful to @jackye1995 for his excellent work. I think Jack summarized this PR from the background, implementation, benefits very well in this [document](https://docs.google.com/document/d/18N-xwZasXLNEl2407xT1oD08Mv9Kd3p0xMX7ZJFyV20/edit?usp=sharing). And Jack proposed two approach in the document to improve the defects of this PR in CDC case. I would like to make some additions and summaries to this document, and continue to discuss the approach about how to deal with the equality deletes compaction in CDC case at here. @rdblue @openinx
@stevenzwu @kbendick

At first, I want to discuss the effectiveness of current implementation of streaming rewrite(apply delta deletes to delta data). We can measure this effectiveness from the type of data stream and whether the table partition type. For the data stream type, we can be divided into append stream and update stream(include upsert and retract/cdc stream). For the table partition type, we can devided into partitioned and unpartitioned, and partitioned can be further divided into partitions that contain time and partitions that do not contain time. For effectiveness, we use `o` to indicate the streaming rewrite is effective that is mean the streaming rewrite can do the same and good as the `RewriteDataFiles` action and user not need to run `RewriteDataFiles` in addition, and we use `x` to indicate the streaming rewrite is not very effective that is mean the streaming rewrite can not do well as the `RewriteDataFiles` action and user still need to run `RewriteDataFiles` in addition. Finally,
we can get the matrix as follow:
| unpartitioned | partitioned without time | partitioned with time
:--: | :--: | :--: | :--:
append stream | o | o | o
upsert stream | x | x | o
retract stream | x | x | o

For append stream, streaming rewrite can work well in all partition type because there are no delete files and we only need to bin pack data files. For update stream(upsert and retract stream), streaming rewrite can work well with table of partitioned with time. Because the result will become certain over time, such as we have an day and hour partition table and T partition result will be certain in T+1 moment, so we can rewrite T partition after T. However, streaming rewrite can not work well with table of partitioned without time, as Jack mentioned in the doc, the high seqNum equality delete file still in storage and when we read the table, they will still apply to the low seqNum data files even if this data file has been rewritten before. So, I agree with Jack that we should first find a way to deal with the remaining equality deletes.

And then, I want to discuss the approach about how to deal with the remaining equality deletes. Jack list two approach in the doc. The approach A is gathering the equality delete files and then perform a table scan to rewrite the affected data files. And the approach B is to convert the equality deletes to position deletes and then perform a table scan to apply the position deletes and rewrite the affected data files. Both approach require extra table scan and data rewrite, I think these approach is work, but will be complicated and we need to reconsider the cost and benefit.

Finally, I think we can have approach C that we can simply rewrite the delta equality delete files to the new one and replace them by [`rewriteFiles`](https://github.com/apache/iceberg/blob/d85a2b25cfa453d211043cf0868dd18de9827808/api/src/main/java/org/apache/iceberg/RewriteFiles.java#L63) . As I said above, currently impelemention of streaming rewrite can not work well with table of partitioned without time because the high seqNum equality delete files(the delta equality delete files) still in storage, and we know the streaming rewrite can apply all deletes to data files. Therefore, if we can rewrite these delta delete files to a new one and bringing them from older seqNum to a high seqNum(the latest commit seqNum), and then remove these delete files from latest snapthot, we can keep the number of equaliy delete fies at a low and same as the number of data files. In this approach, we no need extra table scan and data rewrite, and we remove unnecessary equality delete files and av
oid apply them to the rewritten data files.

--
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 #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

rdblue commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962068331

@jackye1995, thanks for taking the time to write up your thoughts. That doc is really helpful.

I had previously been thinking about this mostly in terms of an append stream and v1 tables. Like @Reo-LEI notes, inline compaction for append streams is safe (for v1 tables) and is a good idea if you want frequent checkpoints but don't want a ton of small files. For v1 tables, we know that there won't be equality or position delete files, so we can easily compact. For v2 tables, append streams are pretty much like CDC streams because we may have concurrent writers adding equality or position delete files.

That brings us back to Jack's points about adapting this idea to CDC streams. First, equality deletes create sequence number boundaries that make it difficult to compact. But I think there are still some use cases where this is valuable even if we only plan on compacting within a sequence number. Write tasks necessarily align with partitions, so compacting across tasks may still be valuable. For example, CDC writes to an unpartitioned table from multiple tasks will create a data file per task (per checkpoint) that is probably not full size.

While there may be multiple files per partition, I think that the larger use case is compacting across checkpoints and that will require addressing compaction across sequence numbers. Let's assume that we have a fairly regular CDC stream for a bucketed table so that the commit for each bucket has one equality delete and one data file per checkpoint, and optionally a position delete file for the data file. That's the worst case that Jack describes, where there is nothing to compact within each sequence number except position deletes. Here are the files committed by sequence number for just one bucket (bucket_id = 0):

| Type | Seq 107 | Seq 106 | Seq 105 | Seq 104 | Seq 103 |
| -- | -- | -- | -- | -- | -- |
| data | data-107.parquet | data-106.parquet | data-105.parquet | data-104.parquet | data-103.parquet |
| eq-deletes | eq-107.parquet | eq-106.parquet | eq-105.parquet | eq-104.parquet | eq-103.parquet |
| pos-deletes | | pos-106.parquet | | | pos-103.parquet |

I think that we actually can do some compaction in this situation. Say we want to compact `data-103.parquet` through `data-107.parquet`. Then we need to apply all newer equality deletes to each data file:

| File | Delete files to apply |
| -- | -- |
| data-103.parquet | eq-{104,105,106,107}.parquet, pos-103.parquet |
| data-104.parquet | eq-{105,106,107}.parquet |
| data-105.parquet | eq-{106,107}.parquet |
| data-106.parquet | eq-107.parquet, pos-106.parquet |
| data-107.parquet | |

The position delete files can be removed because they only reference data-103.parquet and data-106.parquet. The equality deletes must remain in the table in case they deleted data in other files. The new compacted data file should be data-107-compacted.parquet and should be committed at sequence number 107 so that future equality deletes are still applied correctly.

Another thing to keep in mind is that we may have equality and position deletes coming from concurrent writers. But in this case all we would need to do is update the task emitting committed data files to emit all delete files, even those from concurrent writers. For example, maybe sequence number 5 was written by a concurrent commit. Then the delete files for that commit, eq-105.parquet, should be used. But the data file should not be part of the compaction (it would be compacted by a parallel Flink job).

I think that this compaction strategy would address Jack's concern and is a safe way to compact across sequence numbers with equality deletes. The key idea is that we leave the equality delete files in the table so we don't have to worry about existing data files that aren't part of the stream we are compacting.

I think that what I'm proposing is similar to @Reo-LEI's comment, except that it limits the scope of the compaction differently. Using time-based partitions would make us reasonably confident that the compaction is targeted, but the drawback is that we would still need to plan a scan, and it may still be impacted by concurrent deletes. But my strategy of keeping the compaction scoped to just files written by one Flink job and incrementally including delete files is a clean way to handle the issues and ensure that compaction won't conflict with one another.

--
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961753691


   @openinx Thanks for your review! I have been resolve the conflict, please go ahead. 😄 


-- 
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] kbendick commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

kbendick commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r743390878



##########
File path: core/src/main/java/org/apache/iceberg/StaticDataTableScan.java
##########
@@ -0,0 +1,125 @@
+/*
+ * 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;
+
+import java.util.List;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+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.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StaticDataTableScan extends DataTableScan {
+  private static final Logger LOG = LoggerFactory.getLogger(StaticDataTableScan.class);
+
+  private static final long DUMMY_SNAPSHOT_ID = 0L;
+
+  private final List<ManifestFile> dataManifests;
+  private final List<ManifestFile> deleteManifests;
+
+  private StaticDataTableScan(Table table,
+                              TableOperations ops) {
+    super(ops, table);
+    this.dataManifests = Lists.newArrayList();
+    this.deleteManifests = Lists.newArrayList();
+  }
+
+  @Override
+  public TableScan useSnapshot(long scanSnapshotId) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table using scan snapshot id %s is not supported", scanSnapshotId));
+  }
+
+  @Override
+  public TableScan asOfTime(long timestampMillis) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table as of time %s is not supported", timestampMillis));
+  }
+
+  @Override
+  public TableScan appendsBetween(long fromSnapshotId, long toSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public TableScan appendsAfter(long fromSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public Snapshot snapshot() {
+    return null;
+  }
+
+  public TableScan scan(Iterable<ManifestFile> newManifests) {
+    for (ManifestFile manifestFile : newManifests) {
+      Preconditions.checkArgument(manifestFile != null, "Cannot scan a null manifest file");
+      switch (manifestFile.content()) {
+        case DATA:
+          dataManifests.add(manifestFile);
+          break;
+        case DELETES:
+          deleteManifests.add(manifestFile);
+          break;
+        default: throw new IllegalArgumentException("Unknown ManifestContent type: " + manifestFile.content());
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public CloseableIterable<FileScanTask> planFiles() {
+    LOG.info("Scanning table {} partially with filter {}", table(), context().rowFilter());
+
+    Listeners.notifyAll(new ScanEvent(table().name(), DUMMY_SNAPSHOT_ID, context().rowFilter(), schema()));

Review comment:
       WAP is write audit publish. I’m not super familiar with all of the details, but essentially, users can stage a commit or set of commits, then run whatever audit or some sort of validation on them, and then “publish” or finally commit the transaction to the main table.




-- 
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] jackye1995 commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r743390226



##########
File path: api/src/main/java/org/apache/iceberg/types/TypeUtil.java
##########
@@ -282,10 +282,10 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
 
     switch (from.typeId()) {
       case INTEGER:
-        return to == Types.LongType.get();
+        return to.equals(Types.LongType.get());

Review comment:
       I think we use `StructLikeWrapper` to compare any `StructLike` such as `PartitionData`, for example: https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java#L396-L398 




-- 
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] kbendick commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

kbendick commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735258419



##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
##########
@@ -115,10 +129,12 @@ public void setupTable() throws IOException {
 
   @Test
   public void testCommitTxnWithoutDataFiles() throws Exception {
+    Assume.assumeTrue("Only support unpartitioned table.", !partitioned);

Review comment:
       Nit: can you further clarify what is not supported in the assumption statements? Like the thing causing the assumption to be violated is that the table is partitioned, but can you add what is being violated? It seems as though presently the whole process doesn't support partitioned tables, is that correct?




-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735329129



##########
File path: api/src/main/java/org/apache/iceberg/types/TypeUtil.java
##########
@@ -282,10 +282,10 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
 
     switch (from.typeId()) {
       case INTEGER:
-        return to == Types.LongType.get();
+        return to.equals(Types.LongType.get());

Review comment:
       The root cause is when comparing two [PartitionData](https://github.com/apache/iceberg/blob/24896d84c586c6339752a69773409219c0c0ac02/core/src/main/java/org/apache/iceberg/PartitionData.java#L188) for equality we need to compare all type of partition. Currently, we compare whether the two `PrimitiveType` are equal, we compare the singleton PrimitiveType instance. But once this object is serialized and deserialized, the new instance will no longer equal to the old instance, even they are the same type.
   
   For fix this probelm, I add [equal](https://github.com/apache/iceberg/blob/6439bbb2fe8ff84d6bcf2d26f764a61b6e04cf8c/api/src/main/java/org/apache/iceberg/types/Type.java#L117) and [hashcode](https://github.com/apache/iceberg/blob/6439bbb2fe8ff84d6bcf2d26f764a61b6e04cf8c/api/src/main/java/org/apache/iceberg/types/Type.java#L128) method for `PrimitiveType`. And I found that will break some [style checks](https://github.com/apache/iceberg/runs/3953641733?check_suite_focus=true#step:4:313), so I need to chang this.




-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735371033



##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
##########
@@ -616,14 +651,15 @@ public void testFlinkManifests() throws Exception {
   @Test
   public void testDeleteFiles() throws Exception {
     Assume.assumeFalse("Only support equality-delete in format v2.", formatVersion < 2);
+    Assume.assumeTrue("Only support unpartitioned table.", !partitioned);

Review comment:
       I think `assumeTrue` will make it easier to understand. If you want to change to `assumeFalse`, I think that is ok.




-- 
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 #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

rdblue commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962068331

@jackye1995, thanks for taking the time to write up your thoughts. That doc is really helpful.

--
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735401466



##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/DataFileGroup.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.flink.sink;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.KryoSerializable;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class DataFileGroup implements KryoSerializable {
+  private static final Logger LOG = LoggerFactory.getLogger(DataFileGroup.class);
+
+  private long latestSequenceNumber;
+  private long lastSnapshotId;
+  private int filesCount;
+  private long filesSize;
+  private transient List<DeltaManifests> manifestsList;
+
+  DataFileGroup() {
+    this.latestSequenceNumber = 0;
+    this.lastSnapshotId = 0;
+    this.filesCount = 0;
+    this.filesSize = 0;
+    this.manifestsList = Lists.newArrayList();
+  }
+
+  long latestSequenceNumber() {
+    return latestSequenceNumber;
+  }
+
+  long latestSnapshotId() {
+    return lastSnapshotId;
+  }
+
+  int filesCount() {
+    return filesCount;
+  }
+
+  long filesSize() {
+    return filesSize;
+  }
+
+  List<DeltaManifests> manifestsList() {
+    return manifestsList;
+  }
+
+  Iterable<ManifestFile> manifestFiles() {
+    return Iterables.concat(Lists.transform(manifestsList, DeltaManifests::manifests));
+  }
+
+  void append(int dataFilesCount, long dataFliesSize, long snapshotId, long sequenceNumber,
+              DeltaManifests deltaManifests) throws IOException {
+    if (deltaManifests == null || deltaManifests.manifests().isEmpty()) {
+      return;
+    }
+
+    if (sequenceNumber > latestSequenceNumber) {
+      latestSequenceNumber = sequenceNumber;
+      lastSnapshotId = snapshotId;
+    }
+
+    filesCount += dataFilesCount;
+    filesSize += dataFliesSize;
+    manifestsList.add(deltaManifests);
+  }
+
+  @Override
+  public void write(Kryo kryo, Output output) {
+    try {
+      output.writeLong(latestSequenceNumber);
+      output.writeLong(lastSnapshotId);
+      output.writeInt(filesCount);
+      output.writeLong(filesSize);
+
+      int size = manifestsList.size();
+      output.writeInt(size);
+      for (DeltaManifests manifests : manifestsList) {
+        byte[] data = SimpleVersionedSerialization.writeVersionAndSerialize(
+            DeltaManifestsSerializer.INSTANCE, manifests);
+        output.writeInt(data.length);
+        output.writeBytes(data);
+      }
+    } catch (IOException e) {
+      LOG.error("Cannot serialize data file group by kryo.", e);

Review comment:
       I simulated this situation, if the `IOException` has been raised before, the `DataFileGroup.read` will raise a `KryoException` because the input byte buff lose a part of data. So from the result, user will not lose data, but user would not recover flink job by this checkpoint.  
   
   But Kryo no allow raise error when write/read in `KryoSerializable`, so I can't interrupt this progress when the exception is happend. This is also a point that bothers me a lot.




-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735368383



##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
##########
@@ -115,10 +129,12 @@ public void setupTable() throws IOException {
 
   @Test
   public void testCommitTxnWithoutDataFiles() throws Exception {
+    Assume.assumeTrue("Only support unpartitioned table.", !partitioned);

Review comment:
       Actually, the `IcebergFilesCommitter` will do the same work for partition and unpartition table before emit `CommitResult`. So we only need to test `IcebergFilesCommitter` by partition or unpartition table. And the whole test are bese on unpartition table before. But this PR need to emit the `CommitResult` which is group by partition, So I add the `partitioned` param and add this assume to keep the original test. 
   
   Maybe we could change to `Only test for unpartitioned table.`? What do you think?
   But the `partitioned` param doesn't seem necessary now, I will remove the partitioned




-- 
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] kbendick commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

kbendick commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735258825



##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
##########
@@ -616,14 +651,15 @@ public void testFlinkManifests() throws Exception {
   @Test
   public void testDeleteFiles() throws Exception {
     Assume.assumeFalse("Only support equality-delete in format v2.", formatVersion < 2);
+    Assume.assumeTrue("Only support unpartitioned table.", !partitioned);

Review comment:
       Nit: why the change from assumeFalse?




-- 
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 #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

openinx commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961691841






-- 
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] Initial-neko commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Initial-neko commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-1072079818


   a quick question, can this pr be used in flink1.13.1?


-- 
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] hameizi commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

hameizi commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-989684278


   > The key idea is that we leave the equality delete files in the table so we don't have to worry about existing data files that aren't part of the stream we are compacting.
   
   @rdblue If there is too much equality delete files will cause memory pressure when read table.


-- 
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-1025491259

I finish the reactor of this PR according the [comment ](https://github.com/apache/iceberg/pull/3323#issuecomment-962068331) of @rdblue and have been updated the description of this PR.

Briefly explain the change, I add the `IcebergRewriteTaskEmitter` to collect all data files and delete files which are committed by the flink job and collect all eq-delete files which are committed by other writers. Emit `CombinedScanTask` to `IcebergStreamRewriter` to make same partition rewrite file group can be rewritten in paralle. Group rewrite results by starting snapshot id and partition and commit them in batch.

I test this in our prod env and has been running for a while, and that is work for v1 and v2 table. And for the time-based partitioned v1 table, we can use streaming rewrite to replace the batct rewrite action.

@rdblue @jackye1995 @openinx @stevenzwu @kbendick Could you please take another look when you free?

--
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] jackye1995 commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961692478


   had a lot of thinking around this PR, I decide to write down in a doc, @openinx @rdblue @stevenzwu please let me know what you think!
   
   https://docs.google.com/document/d/18N-xwZasXLNEl2407xT1oD08Mv9Kd3p0xMX7ZJFyV20/edit?usp=sharing


-- 
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] Reo-LEI edited a comment on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI edited a comment on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962011180

--
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735329129



##########
File path: api/src/main/java/org/apache/iceberg/types/TypeUtil.java
##########
@@ -282,10 +282,10 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
 
     switch (from.typeId()) {
       case INTEGER:
-        return to == Types.LongType.get();
+        return to.equals(Types.LongType.get());

Review comment:
       The root cause is when comparing two [PartitionData](https://github.com/apache/iceberg/blob/24896d84c586c6339752a69773409219c0c0ac02/core/src/main/java/org/apache/iceberg/PartitionData.java#L188) for equality, we need to compare all type of partition. Currently, we compare whether the two `PrimitiveType` are equal, we compare the singleton PrimitiveType instance, once this object is serialized and deserialized, the new instance will no longer equal to the old instance, even they are the same type.
   
   For fix this probelm, I add [equal](https://github.com/apache/iceberg/blob/6439bbb2fe8ff84d6bcf2d26f764a61b6e04cf8c/api/src/main/java/org/apache/iceberg/types/Type.java#L117) and [hashcode](https://github.com/apache/iceberg/blob/6439bbb2fe8ff84d6bcf2d26f764a61b6e04cf8c/api/src/main/java/org/apache/iceberg/types/Type.java#L128) method for `PrimitiveType`. And I found that will break some [style checks](https://github.com/apache/iceberg/runs/3953641733?check_suite_focus=true#step:4:313), so I need to chang this.




-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735401466



##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/DataFileGroup.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.flink.sink;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.KryoSerializable;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class DataFileGroup implements KryoSerializable {
+  private static final Logger LOG = LoggerFactory.getLogger(DataFileGroup.class);
+
+  private long latestSequenceNumber;
+  private long lastSnapshotId;
+  private int filesCount;
+  private long filesSize;
+  private transient List<DeltaManifests> manifestsList;
+
+  DataFileGroup() {
+    this.latestSequenceNumber = 0;
+    this.lastSnapshotId = 0;
+    this.filesCount = 0;
+    this.filesSize = 0;
+    this.manifestsList = Lists.newArrayList();
+  }
+
+  long latestSequenceNumber() {
+    return latestSequenceNumber;
+  }
+
+  long latestSnapshotId() {
+    return lastSnapshotId;
+  }
+
+  int filesCount() {
+    return filesCount;
+  }
+
+  long filesSize() {
+    return filesSize;
+  }
+
+  List<DeltaManifests> manifestsList() {
+    return manifestsList;
+  }
+
+  Iterable<ManifestFile> manifestFiles() {
+    return Iterables.concat(Lists.transform(manifestsList, DeltaManifests::manifests));
+  }
+
+  void append(int dataFilesCount, long dataFliesSize, long snapshotId, long sequenceNumber,
+              DeltaManifests deltaManifests) throws IOException {
+    if (deltaManifests == null || deltaManifests.manifests().isEmpty()) {
+      return;
+    }
+
+    if (sequenceNumber > latestSequenceNumber) {
+      latestSequenceNumber = sequenceNumber;
+      lastSnapshotId = snapshotId;
+    }
+
+    filesCount += dataFilesCount;
+    filesSize += dataFliesSize;
+    manifestsList.add(deltaManifests);
+  }
+
+  @Override
+  public void write(Kryo kryo, Output output) {
+    try {
+      output.writeLong(latestSequenceNumber);
+      output.writeLong(lastSnapshotId);
+      output.writeInt(filesCount);
+      output.writeLong(filesSize);
+
+      int size = manifestsList.size();
+      output.writeInt(size);
+      for (DeltaManifests manifests : manifestsList) {
+        byte[] data = SimpleVersionedSerialization.writeVersionAndSerialize(
+            DeltaManifestsSerializer.INSTANCE, manifests);
+        output.writeInt(data.length);
+        output.writeBytes(data);
+      }
+    } catch (IOException e) {
+      LOG.error("Cannot serialize data file group by kryo.", e);

Review comment:
       I simulated this situation, if the `IOException` has been raised before, the `DataFileGroup.read` will raise a `KryoException` because the input byte buff lose a part of data. So from the result, user will not lose data, but user would not recover flink job by this checkpoint.  
   
   But Kryo no allow raise error when write/read in `KryoSerializable`, so I can't interrupt this progress when the exception is happend. This is also a point that bothers me a lot. Do you have any suggestions for this? 




-- 
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 #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

rdblue commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962068331

@jackye1995, thanks for taking the time to write up your thoughts. That doc is really helpful.

--
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] kbendick commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

kbendick commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r743390878



##########
File path: core/src/main/java/org/apache/iceberg/StaticDataTableScan.java
##########
@@ -0,0 +1,125 @@
+/*
+ * 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;
+
+import java.util.List;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+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.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StaticDataTableScan extends DataTableScan {
+  private static final Logger LOG = LoggerFactory.getLogger(StaticDataTableScan.class);
+
+  private static final long DUMMY_SNAPSHOT_ID = 0L;
+
+  private final List<ManifestFile> dataManifests;
+  private final List<ManifestFile> deleteManifests;
+
+  private StaticDataTableScan(Table table,
+                              TableOperations ops) {
+    super(ops, table);
+    this.dataManifests = Lists.newArrayList();
+    this.deleteManifests = Lists.newArrayList();
+  }
+
+  @Override
+  public TableScan useSnapshot(long scanSnapshotId) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table using scan snapshot id %s is not supported", scanSnapshotId));
+  }
+
+  @Override
+  public TableScan asOfTime(long timestampMillis) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table as of time %s is not supported", timestampMillis));
+  }
+
+  @Override
+  public TableScan appendsBetween(long fromSnapshotId, long toSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public TableScan appendsAfter(long fromSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public Snapshot snapshot() {
+    return null;
+  }
+
+  public TableScan scan(Iterable<ManifestFile> newManifests) {
+    for (ManifestFile manifestFile : newManifests) {
+      Preconditions.checkArgument(manifestFile != null, "Cannot scan a null manifest file");
+      switch (manifestFile.content()) {
+        case DATA:
+          dataManifests.add(manifestFile);
+          break;
+        case DELETES:
+          deleteManifests.add(manifestFile);
+          break;
+        default: throw new IllegalArgumentException("Unknown ManifestContent type: " + manifestFile.content());
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public CloseableIterable<FileScanTask> planFiles() {
+    LOG.info("Scanning table {} partially with filter {}", table(), context().rowFilter());
+
+    Listeners.notifyAll(new ScanEvent(table().name(), DUMMY_SNAPSHOT_ID, context().rowFilter(), schema()));

Review comment:
       WAP is write audit publish. I’m not super familiar with all of the details, but essentially, users can stage a commit or set of commits, then run whatever audit or some sort of validation on them, and then “publish” or finally commit the transaction to the main table.




-- 
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 #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

openinx commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961691841


   @Reo-LEI would you mind to fix this conflicts ?  I will plan to go through the whole PR, thanks for the contribution !


-- 
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] kbendick commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

kbendick commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r743390878



##########
File path: core/src/main/java/org/apache/iceberg/StaticDataTableScan.java
##########
@@ -0,0 +1,125 @@
+/*
+ * 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;
+
+import java.util.List;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+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.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StaticDataTableScan extends DataTableScan {
+  private static final Logger LOG = LoggerFactory.getLogger(StaticDataTableScan.class);
+
+  private static final long DUMMY_SNAPSHOT_ID = 0L;
+
+  private final List<ManifestFile> dataManifests;
+  private final List<ManifestFile> deleteManifests;
+
+  private StaticDataTableScan(Table table,
+                              TableOperations ops) {
+    super(ops, table);
+    this.dataManifests = Lists.newArrayList();
+    this.deleteManifests = Lists.newArrayList();
+  }
+
+  @Override
+  public TableScan useSnapshot(long scanSnapshotId) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table using scan snapshot id %s is not supported", scanSnapshotId));
+  }
+
+  @Override
+  public TableScan asOfTime(long timestampMillis) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table as of time %s is not supported", timestampMillis));
+  }
+
+  @Override
+  public TableScan appendsBetween(long fromSnapshotId, long toSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public TableScan appendsAfter(long fromSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public Snapshot snapshot() {
+    return null;
+  }
+
+  public TableScan scan(Iterable<ManifestFile> newManifests) {
+    for (ManifestFile manifestFile : newManifests) {
+      Preconditions.checkArgument(manifestFile != null, "Cannot scan a null manifest file");
+      switch (manifestFile.content()) {
+        case DATA:
+          dataManifests.add(manifestFile);
+          break;
+        case DELETES:
+          deleteManifests.add(manifestFile);
+          break;
+        default: throw new IllegalArgumentException("Unknown ManifestContent type: " + manifestFile.content());
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public CloseableIterable<FileScanTask> planFiles() {
+    LOG.info("Scanning table {} partially with filter {}", table(), context().rowFilter());
+
+    Listeners.notifyAll(new ScanEvent(table().name(), DUMMY_SNAPSHOT_ID, context().rowFilter(), schema()));

Review comment:
       WAP is write audit publish. I’m not super familiar with all of the details, but essentially, users can stage a commit or set of commits, then run whatever audit or some sort of validation on them, and then “publish” or finally commit the transaction to the main table.




-- 
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-950067848


   I this PR is ok, could you take a look of this? 😄 @rdblue @openinx @stevenzwu @jackye1995 @kbendick 


-- 
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-947301754


   I just finish the streaming rewrite function, and I will complete the unittest in these few days. Maybe you can take a quick look and leave some comments, I will keep improve this PR in this week. @rdblue @openinx @stevenzwu @jackye1995 @kbendick  :) 


-- 
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] Reo-LEI edited a comment on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI edited a comment on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962011180

--
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961753691






-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735345230



##########
File path: core/src/main/java/org/apache/iceberg/StaticDataTableScan.java
##########
@@ -0,0 +1,125 @@
+/*
+ * 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;
+
+import java.util.List;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+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.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StaticDataTableScan extends DataTableScan {
+  private static final Logger LOG = LoggerFactory.getLogger(StaticDataTableScan.class);
+
+  private static final long DUMMY_SNAPSHOT_ID = 0L;
+
+  private final List<ManifestFile> dataManifests;
+  private final List<ManifestFile> deleteManifests;
+
+  private StaticDataTableScan(Table table,
+                              TableOperations ops) {
+    super(ops, table);
+    this.dataManifests = Lists.newArrayList();
+    this.deleteManifests = Lists.newArrayList();
+  }
+
+  @Override
+  public TableScan useSnapshot(long scanSnapshotId) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table using scan snapshot id %s is not supported", scanSnapshotId));
+  }
+
+  @Override
+  public TableScan asOfTime(long timestampMillis) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table as of time %s is not supported", timestampMillis));
+  }
+
+  @Override
+  public TableScan appendsBetween(long fromSnapshotId, long toSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public TableScan appendsAfter(long fromSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public Snapshot snapshot() {
+    return null;
+  }
+
+  public TableScan scan(Iterable<ManifestFile> newManifests) {
+    for (ManifestFile manifestFile : newManifests) {
+      Preconditions.checkArgument(manifestFile != null, "Cannot scan a null manifest file");
+      switch (manifestFile.content()) {
+        case DATA:
+          dataManifests.add(manifestFile);
+          break;
+        case DELETES:
+          deleteManifests.add(manifestFile);
+          break;
+        default: throw new IllegalArgumentException("Unknown ManifestContent type: " + manifestFile.content());
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public CloseableIterable<FileScanTask> planFiles() {
+    LOG.info("Scanning table {} partially with filter {}", table(), context().rowFilter());
+
+    Listeners.notifyAll(new ScanEvent(table().name(), DUMMY_SNAPSHOT_ID, context().rowFilter(), schema()));

Review comment:
       Here I use `DUMMY_SNAPSHOT_ID` just because `StaticDataTableScan` only scan the specific manifests but not scan any snapshot. If this notify will affect other feature, I think we could remove this simply.




-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735350470



##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/CommitResult.java
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.flink.sink;
+
+import java.io.Serializable;
+import java.util.Collection;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.WriteResult;
+
+class CommitResult implements Serializable {
+
+  private final long snapshotId;
+  private final long sequenceNumber;
+  private final int specId;
+  private final StructLike partition;
+  private final WriteResult writeResult;
+
+  private CommitResult(long snapshotId,
+                       long sequenceNumber,
+                       int specId,
+                       StructLike partition,
+                       WriteResult writeResult) {
+    this.snapshotId = snapshotId;
+    this.sequenceNumber = sequenceNumber;
+    this.specId = specId;
+    this.partition = partition;
+    this.writeResult = writeResult;
+  }
+
+  long snapshotId() {
+    return snapshotId;
+  }
+
+  long sequenceNumber() {
+    return sequenceNumber;
+  }
+
+  public int specId() {
+    return specId;
+  }
+
+  StructLike partition() {
+    return partition;
+  }
+
+  WriteResult writeResult() {
+    return writeResult;
+  }
+
+  static Builder builder(long snapshotId, long sequenceNumber) {
+    return new Builder(snapshotId, sequenceNumber);
+  }
+
+  static class Builder {
+
+    private final long snapshotId;
+    private final long sequenceNumber;
+    private int specId;
+    private StructLike partition;
+    private final WriteResult.Builder writeResult;
+
+    private Builder(long snapshotId, long sequenceNumber) {
+      this.snapshotId = snapshotId;
+      this.sequenceNumber = sequenceNumber;
+      this.specId = 0;
+      this.partition = null;
+      this.writeResult = WriteResult.builder();
+    }
+
+    Builder partition(int newSpecId, StructLike newPartition) {
+      this.specId = newSpecId;
+      this.partition = newPartition;
+      return this;
+    }
+
+    Builder addDataFile(Collection<DataFile> files) {
+      writeResult.addDataFiles(files);
+      return this;
+    }
+
+    Builder addDeleteFile(Collection<DeleteFile> files) {
+      writeResult.addDeleteFiles(files);
+      return this;
+    }
+
+    Builder addReferencedDataFile(Collection<CharSequence> files) {
+      writeResult.addReferencedDataFiles(files);

Review comment:
       Those only the pos-delete file referenced data files path location, and that are [collect](https://github.com/apache/iceberg/blob/6439bbb2fe8ff84d6bcf2d26f764a61b6e04cf8c/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java#L365) from committed `WriteResult`.




-- 
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-1072162035


   > a quick question, can this pr be used in flink1.13.1?
   
   I have been port this featuer to flink 1.13 in branch ‘flink-streaming-rewrite-for-flink0.13’ before, but there have some conflicts need to be resloved, maybe you can use that. @Initial-neko 


-- 
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 #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

stevenzwu commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735248351



##########
File path: api/src/main/java/org/apache/iceberg/types/TypeUtil.java
##########
@@ -282,10 +282,10 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
 
     switch (from.typeId()) {
       case INTEGER:
-        return to == Types.LongType.get();
+        return to.equals(Types.LongType.get());

Review comment:
       why is this change necessary?




-- 
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962011180

For append stream, streaming rewrite can work well in all partition type because there are no delete files and just bin pack data files. For update stream(upsert and retract stream), streaming rewrite can work well with table of partitioned with time. Because the result will become certain over time, such as we have an day and hour partition table and T partition result will be certain in T+1 moment, so we can rewrite T partition after T. However, streaming rewrite can not work well with table of partitioned without time, as Jack mentioned in the doc, the high seqNum equality delete file still in storage and when we read the table, they will still apply to the low seqNum data files even if this data file has been rewritten before. So, I agree with Jack that we should first find a way to deal with the remaining equality 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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961753691






-- 
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] jackye1995 commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-961692478


   had a lot of thinking around this PR, I decide to write down in a doc, @openinx @rdblue @stevenzwu please let me know what you think!
   
   https://docs.google.com/document/d/18N-xwZasXLNEl2407xT1oD08Mv9Kd3p0xMX7ZJFyV20/edit?usp=sharing


-- 
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] Initial-neko commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Initial-neko commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-1072079818


   a quick question, can this pr be used in flink1.13.1?


-- 
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] Reo-LEI commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-1072162035


   > a quick question, can this pr be used in flink1.13.1?
   
   I have been port this featuer to flink 1.13 in branch ‘flink-streaming-rewrite-for-flink0.13’ before, but there have some conflicts need to be resloved, maybe you can use that. @Initial-neko 


-- 
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] kbendick commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

kbendick commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735250691



##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/CommitResult.java
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.flink.sink;
+
+import java.io.Serializable;
+import java.util.Collection;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.WriteResult;
+
+class CommitResult implements Serializable {
+
+  private final long snapshotId;
+  private final long sequenceNumber;
+  private final int specId;
+  private final StructLike partition;
+  private final WriteResult writeResult;
+
+  private CommitResult(long snapshotId,
+                       long sequenceNumber,
+                       int specId,
+                       StructLike partition,
+                       WriteResult writeResult) {
+    this.snapshotId = snapshotId;
+    this.sequenceNumber = sequenceNumber;
+    this.specId = specId;
+    this.partition = partition;
+    this.writeResult = writeResult;
+  }
+
+  long snapshotId() {
+    return snapshotId;
+  }
+
+  long sequenceNumber() {
+    return sequenceNumber;
+  }
+
+  public int specId() {
+    return specId;
+  }
+
+  StructLike partition() {
+    return partition;
+  }
+
+  WriteResult writeResult() {
+    return writeResult;
+  }
+
+  static Builder builder(long snapshotId, long sequenceNumber) {
+    return new Builder(snapshotId, sequenceNumber);
+  }
+
+  static class Builder {
+
+    private final long snapshotId;
+    private final long sequenceNumber;
+    private int specId;
+    private StructLike partition;
+    private final WriteResult.Builder writeResult;
+
+    private Builder(long snapshotId, long sequenceNumber) {
+      this.snapshotId = snapshotId;
+      this.sequenceNumber = sequenceNumber;
+      this.specId = 0;
+      this.partition = null;
+      this.writeResult = WriteResult.builder();
+    }
+
+    Builder partition(int newSpecId, StructLike newPartition) {
+      this.specId = newSpecId;
+      this.partition = newPartition;
+      return this;
+    }
+
+    Builder addDataFile(Collection<DataFile> files) {
+      writeResult.addDataFiles(files);
+      return this;
+    }
+
+    Builder addDeleteFile(Collection<DeleteFile> files) {
+      writeResult.addDeleteFiles(files);
+      return this;
+    }
+
+    Builder addReferencedDataFile(Collection<CharSequence> files) {
+      writeResult.addReferencedDataFiles(files);

Review comment:
       Clarifying question: These are data files for which rows in the delete dataset are found?

##########
File path: core/src/main/java/org/apache/iceberg/StaticDataTableScan.java
##########
@@ -0,0 +1,125 @@
+/*
+ * 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;
+
+import java.util.List;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+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.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StaticDataTableScan extends DataTableScan {
+  private static final Logger LOG = LoggerFactory.getLogger(StaticDataTableScan.class);
+
+  private static final long DUMMY_SNAPSHOT_ID = 0L;
+
+  private final List<ManifestFile> dataManifests;
+  private final List<ManifestFile> deleteManifests;
+
+  private StaticDataTableScan(Table table,
+                              TableOperations ops) {
+    super(ops, table);
+    this.dataManifests = Lists.newArrayList();
+    this.deleteManifests = Lists.newArrayList();
+  }
+
+  @Override
+  public TableScan useSnapshot(long scanSnapshotId) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table using scan snapshot id %s is not supported", scanSnapshotId));
+  }
+
+  @Override
+  public TableScan asOfTime(long timestampMillis) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table as of time %s is not supported", timestampMillis));
+  }
+
+  @Override
+  public TableScan appendsBetween(long fromSnapshotId, long toSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public TableScan appendsAfter(long fromSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public Snapshot snapshot() {
+    return null;
+  }
+
+  public TableScan scan(Iterable<ManifestFile> newManifests) {
+    for (ManifestFile manifestFile : newManifests) {
+      Preconditions.checkArgument(manifestFile != null, "Cannot scan a null manifest file");
+      switch (manifestFile.content()) {
+        case DATA:
+          dataManifests.add(manifestFile);
+          break;
+        case DELETES:
+          deleteManifests.add(manifestFile);
+          break;
+        default: throw new IllegalArgumentException("Unknown ManifestContent type: " + manifestFile.content());
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public CloseableIterable<FileScanTask> planFiles() {
+    LOG.info("Scanning table {} partially with filter {}", table(), context().rowFilter());
+
+    Listeners.notifyAll(new ScanEvent(table().name(), DUMMY_SNAPSHOT_ID, context().rowFilter(), schema()));

Review comment:
       Is the `dummySnapshotId` normal for WAP events? I'm admittedly not as familiar with them. Will this break the WAP ability?

##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/DataFileGroup.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.flink.sink;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.KryoSerializable;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class DataFileGroup implements KryoSerializable {
+  private static final Logger LOG = LoggerFactory.getLogger(DataFileGroup.class);
+
+  private long latestSequenceNumber;
+  private long lastSnapshotId;
+  private int filesCount;
+  private long filesSize;
+  private transient List<DeltaManifests> manifestsList;
+
+  DataFileGroup() {
+    this.latestSequenceNumber = 0;
+    this.lastSnapshotId = 0;
+    this.filesCount = 0;
+    this.filesSize = 0;
+    this.manifestsList = Lists.newArrayList();
+  }
+
+  long latestSequenceNumber() {
+    return latestSequenceNumber;
+  }
+
+  long latestSnapshotId() {
+    return lastSnapshotId;
+  }
+
+  int filesCount() {
+    return filesCount;
+  }
+
+  long filesSize() {
+    return filesSize;
+  }
+
+  List<DeltaManifests> manifestsList() {
+    return manifestsList;
+  }
+
+  Iterable<ManifestFile> manifestFiles() {
+    return Iterables.concat(Lists.transform(manifestsList, DeltaManifests::manifests));
+  }
+
+  void append(int dataFilesCount, long dataFliesSize, long snapshotId, long sequenceNumber,
+              DeltaManifests deltaManifests) throws IOException {
+    if (deltaManifests == null || deltaManifests.manifests().isEmpty()) {
+      return;
+    }
+
+    if (sequenceNumber > latestSequenceNumber) {
+      latestSequenceNumber = sequenceNumber;
+      lastSnapshotId = snapshotId;
+    }
+
+    filesCount += dataFilesCount;
+    filesSize += dataFliesSize;
+    manifestsList.add(deltaManifests);
+  }
+
+  @Override
+  public void write(Kryo kryo, Output output) {
+    try {
+      output.writeLong(latestSequenceNumber);
+      output.writeLong(lastSnapshotId);
+      output.writeInt(filesCount);
+      output.writeLong(filesSize);
+
+      int size = manifestsList.size();
+      output.writeInt(size);
+      for (DeltaManifests manifests : manifestsList) {
+        byte[] data = SimpleVersionedSerialization.writeVersionAndSerialize(
+            DeltaManifestsSerializer.INSTANCE, manifests);
+        output.writeInt(data.length);
+        output.writeBytes(data);
+      }
+    } catch (IOException e) {
+      LOG.error("Cannot serialize data file group by kryo.", e);

Review comment:
       If this is hit (which I know shouldn't be expected), will the user possibly lose data?




-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735340439



##########
File path: core/src/main/java/org/apache/iceberg/StaticDataTableScan.java
##########
@@ -0,0 +1,125 @@
+/*
+ * 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;
+
+import java.util.List;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+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.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StaticDataTableScan extends DataTableScan {
+  private static final Logger LOG = LoggerFactory.getLogger(StaticDataTableScan.class);
+
+  private static final long DUMMY_SNAPSHOT_ID = 0L;
+
+  private final List<ManifestFile> dataManifests;
+  private final List<ManifestFile> deleteManifests;
+
+  private StaticDataTableScan(Table table,
+                              TableOperations ops) {
+    super(ops, table);
+    this.dataManifests = Lists.newArrayList();
+    this.deleteManifests = Lists.newArrayList();
+  }
+
+  @Override
+  public TableScan useSnapshot(long scanSnapshotId) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table using scan snapshot id %s is not supported", scanSnapshotId));
+  }
+
+  @Override
+  public TableScan asOfTime(long timestampMillis) {
+    throw new UnsupportedOperationException(String.format(
+        "Scan table as of time %s is not supported", timestampMillis));
+  }
+
+  @Override
+  public TableScan appendsBetween(long fromSnapshotId, long toSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public TableScan appendsAfter(long fromSnapshotId) {
+    throw new UnsupportedOperationException("Incremental scan is not supported");
+  }
+
+  @Override
+  public Snapshot snapshot() {
+    return null;
+  }
+
+  public TableScan scan(Iterable<ManifestFile> newManifests) {
+    for (ManifestFile manifestFile : newManifests) {
+      Preconditions.checkArgument(manifestFile != null, "Cannot scan a null manifest file");
+      switch (manifestFile.content()) {
+        case DATA:
+          dataManifests.add(manifestFile);
+          break;
+        case DELETES:
+          deleteManifests.add(manifestFile);
+          break;
+        default: throw new IllegalArgumentException("Unknown ManifestContent type: " + manifestFile.content());
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public CloseableIterable<FileScanTask> planFiles() {
+    LOG.info("Scanning table {} partially with filter {}", table(), context().rowFilter());
+
+    Listeners.notifyAll(new ScanEvent(table().name(), DUMMY_SNAPSHOT_ID, context().rowFilter(), schema()));

Review comment:
       Can you please help me understand the ability of WAP? I didn’t know that there was this feature before.




-- 
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] Reo-LEI commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

Reo-LEI commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r735368383



##########
File path: flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
##########
@@ -115,10 +129,12 @@ public void setupTable() throws IOException {
 
   @Test
   public void testCommitTxnWithoutDataFiles() throws Exception {
+    Assume.assumeTrue("Only support unpartitioned table.", !partitioned);

Review comment:
       Actually, the `IcebergFilesCommitter` will do the same work for partition and unpartition table before emit `CommitResult`. So we only need to test `IcebergFilesCommitter` by partition or unpartition table. And the whole test are bese on unpartition table before. But this PR need to emit the `CommitResult` which is group by partition, So I add the `partitioned` param and add this assume to keep the original test. 
   
   Maybe we could change to `Only test for unpartitioned table.`? 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] jackye1995 commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r743390226



##########
File path: api/src/main/java/org/apache/iceberg/types/TypeUtil.java
##########
@@ -282,10 +282,10 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
 
     switch (from.typeId()) {
       case INTEGER:
-        return to == Types.LongType.get();
+        return to.equals(Types.LongType.get());

Review comment:
       I think we use `StructLikeWrapper` to compare any `StructLike` such as `PartitionData`, for example: https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java#L396-L398 




-- 
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] jackye1995 commented on a change in pull request #3323: Flink: Incrementally rewrite data files in streaming.

Posted by GitBox <gi...@apache.org>.

jackye1995 commented on a change in pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#discussion_r743390226



##########
File path: api/src/main/java/org/apache/iceberg/types/TypeUtil.java
##########
@@ -282,10 +282,10 @@ public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
 
     switch (from.typeId()) {
       case INTEGER:
-        return to == Types.LongType.get();
+        return to.equals(Types.LongType.get());

Review comment:
       I think we use `StructLikeWrapper` to compare any `StructLike` such as `PartitionData`, for example: https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java#L396-L398 




-- 
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