You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "leaves12138 (via GitHub)" <gi...@apache.org> on 2023/06/05 07:02:27 UTC
[GitHub] [incubator-paimon] leaves12138 opened a new pull request, #1256: [core] Introduce non-bucket mode for append-only table
leaves12138 opened a new pull request, #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256
[core] add negative bucket mode, separate compaction from appendonly writer, add test for negative bucket table #1224
In paimon core project, add the negative bucket process.
<!-- What is the purpose of the change, or the associated issue -->
### Tests
paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionTest.java
paimon-core/src/test/java/org/apache/paimon/table/sink/CommitMessageSerializerTest.java
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1209744793
##########
paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionWorker.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.append;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/** {@link AppendOnlyFileStoreTable} compact worker. */
+public class AppendOnlyTableCompactionWorker {
+ private final List<CompactionTask> tasks = new ArrayList<>();
+ private final List<CommitMessage> result = new ArrayList<>();
+ private final String commitUser = UUID.randomUUID().toString();
+ private final AppendOnlyFileStoreWrite write;
+ private final Map<BinaryRow, AppendOnlyCompactManager.CompactRewriter> rewriters =
+ new HashMap<>();
+
+ public AppendOnlyTableCompactionWorker(AppendOnlyFileStoreTable table) {
+ this.write = table.store().newWrite(commitUser);
+ }
+
+ public AppendOnlyTableCompactionWorker accept(List<CompactionTask> tasks) {
+ this.tasks.addAll(tasks);
+ return this;
+ }
+
+ public AppendOnlyTableCompactionWorker accept(CompactionTask task) {
+ this.tasks.add(task);
+ return this;
+ }
+
+ public AppendOnlyTableCompactionWorker doCompact() throws Exception {
Review Comment:
return `List<CommitMessage>`.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] leaves12138 commented on pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#issuecomment-1576201498
will be apart be more than one pull request
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] leaves12138 closed pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 closed pull request #1256: [core] Introduce non-bucket mode for append-only table
URL: https://github.com/apache/incubator-paimon/pull/1256
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1212522738
##########
paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinator.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.paimon.append;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.operation.AppendOnlyFileStoreScan;
+import org.apache.paimon.operation.ScanKind;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** {@link AppendOnlyFileStoreTable} compact coordinator. */
+public class AppendOnlyTableCompactionCoordinator {
+ private final SnapshotManager snapshotManager;
+ private final AppendOnlyFileStoreScan scan;
+
+ private Long snapshotId;
+ private final long targetFileSize;
+ private final int minFileNum;
+ private final int maxFileNum;
+
+ private final Map<BinaryRow, PartitionCompactCoordinator> partitionCompactCoordinators =
+ new HashMap<>();
+
+ public AppendOnlyTableCompactionCoordinator(
Review Comment:
Can you modify the `CompactAction` for this non-bucket append-only table?
You can modify topology like `CompactorSourceBuilder`.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1209732511
##########
paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCheckUtils.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.paimon.append;
+
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.Table;
+
+/** Utils for {@link AppendOnlyFileStoreTable}. */
+public class AppendOnlyTableCheckUtils {
Review Comment:
remove this class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1209741514
##########
paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.paimon.append;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.operation.AppendOnlyFileStoreScan;
+import org.apache.paimon.operation.ScanKind;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.SnapshotManager;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.stream.Collectors;
+
+/** {@link AppendOnlyFileStoreTable} compact coordinator. */
+public class AppendOnlyTableCompactionCoordinator {
+ private final SnapshotManager snapshotManager;
+ private final AppendOnlyFileStoreScan scan;
+ private Long snapshotId;
+
+ private final long targetFileSize;
+ private final int minFileNum;
+ private final int maxFileNum;
+
+ private final Map<BinaryRow, PartitionCompactCoordinator> partitionCompactCoordinators =
+ new HashMap<>();
+
+ public AppendOnlyTableCompactionCoordinator(
+ AppendOnlyFileStoreTable table, long targetFileSize, int minFileNum, int maxFileNum) {
+ scan = table.store().newScan();
+ snapshotManager = table.snapshotManager();
+ this.targetFileSize = targetFileSize;
+ this.minFileNum = minFileNum;
+ this.maxFileNum = maxFileNum;
+ snapshotId = snapshotManager.latestSnapshotId();
+
+ if (snapshotId != null) {
+ updateFull();
+ }
+ }
+
+ public void updateRestore() {
+ Long latestSnapshotId = snapshotManager.latestSnapshotId();
+ if (latestSnapshotId == null || latestSnapshotId.equals(snapshotId)) {
+ return;
+ }
+
+ if (snapshotId == null) {
+ snapshotId = latestSnapshotId;
+ updateFull();
+ } else {
+ updateDelta(latestSnapshotId);
+ snapshotId = latestSnapshotId;
+ }
+ }
+
+ private void updateFull() {
+ scan.withSnapshot(snapshotId).plan().files().stream()
+ .collect(Collectors.groupingBy(ManifestEntry::partition))
+ .forEach(
+ (k, v) ->
+ partitionCompactCoordinators
+ .computeIfAbsent(k, PartitionCompactCoordinator::new)
+ .addRestoredFiles(
+ v.stream()
+ .map(ManifestEntry::file)
+ .filter(
+ dataFileMeta ->
+ dataFileMeta.fileSize()
+ < targetFileSize)
+ .collect(Collectors.toList())));
+ }
+
+ private void updateDelta(Long targetSnapshotId) {
Review Comment:
We don't need to `updateDelta`, just update full and store delta files from writers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] leaves12138 closed pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 closed pull request #1256: [core] Introduce non-bucket mode for append-only table
URL: https://github.com/apache/incubator-paimon/pull/1256
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1209736977
##########
paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java:
##########
@@ -130,9 +135,23 @@ public TableWriteImpl<InternalRow> newWrite(String commitUser) {
@Override
public TableWriteImpl<InternalRow> newWrite(
String commitUser, ManifestCacheFilter manifestFilter) {
+ AppendOnlyFileStoreWrite writer = store().newWrite(commitUser);
+ boolean negativeBucket = coreOptions().bucket() == -1;
+ if (negativeBucket) {
+ writer.skipCompaction();
+ writer.fromEmptyWriter();
+ }
return new TableWriteImpl<>(
- store().newWrite(commitUser, manifestFilter),
- new InternalRowKeyAndBucketExtractor(tableSchema),
+ writer,
+ new InternalRowKeyAndBucketExtractor(tableSchema) {
+ @Override
+ public int bucket() {
+ if (negativeBucket) {
Review Comment:
We should have a `BucketProcessor` to deal with all bucket related things for source and sink.
You can finish this in next PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1209733866
##########
paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java:
##########
@@ -141,4 +160,16 @@ public TableWriteImpl<InternalRow> newWrite(
return record.row();
});
}
+
Review Comment:
```
public NonBucketCompaction createNonBucketCompaction() {
1. check bucket is -1
2. create NonBucketCompaction
}
class NonBucketCompaction {
1. createCoordinator
2. createWorker
}
```
##########
paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.paimon.append;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.operation.AppendOnlyFileStoreScan;
+import org.apache.paimon.operation.ScanKind;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.SnapshotManager;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.stream.Collectors;
+
+/** {@link AppendOnlyFileStoreTable} compact coordinator. */
+public class AppendOnlyTableCompactionCoordinator {
+ private final SnapshotManager snapshotManager;
+ private final AppendOnlyFileStoreScan scan;
+ private Long snapshotId;
+
+ private final long targetFileSize;
+ private final int minFileNum;
+ private final int maxFileNum;
+
+ private final Map<BinaryRow, PartitionCompactCoordinator> partitionCompactCoordinators =
+ new HashMap<>();
+
+ public AppendOnlyTableCompactionCoordinator(
+ AppendOnlyFileStoreTable table, long targetFileSize, int minFileNum, int maxFileNum) {
+ scan = table.store().newScan();
+ snapshotManager = table.snapshotManager();
+ this.targetFileSize = targetFileSize;
+ this.minFileNum = minFileNum;
+ this.maxFileNum = maxFileNum;
+ snapshotId = snapshotManager.latestSnapshotId();
+
+ if (snapshotId != null) {
+ updateFull();
+ }
+ }
+
+ public void updateRestore() {
Review Comment:
I cannot image how to use these methods.
##########
paimon-core/src/main/java/org/apache/paimon/append/CompactionTask.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.paimon.append;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.io.NewFilesIncrement;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.SerializationUtils.deserializedBytes;
+import static org.apache.paimon.utils.SerializationUtils.serializeBytes;
+
+/**
+ * Compaction task generated by {@link AppendOnlyTableCompactionCoordinator} and executed be {@link
+ * AppendOnlyTableCompactionWorker}.
+ */
+public class CompactionTask implements CommitMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final ThreadLocal<CommitMessageSerializer> CACHE =
+ ThreadLocal.withInitial(CommitMessageSerializer::new);
+
+ private BinaryRow partition;
+ private transient List<DataFileMeta> newFiles;
+ private transient List<DataFileMeta> restoredFiles;
+ private transient List<DataFileMeta> compactBefore;
+ private transient List<DataFileMeta> compactAfter;
Review Comment:
What is used?
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] Introduce non-bucket mode for append-only table
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1209744282
##########
paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionWorker.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.append;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/** {@link AppendOnlyFileStoreTable} compact worker. */
+public class AppendOnlyTableCompactionWorker {
+ private final List<CompactionTask> tasks = new ArrayList<>();
+ private final List<CommitMessage> result = new ArrayList<>();
+ private final String commitUser = UUID.randomUUID().toString();
Review Comment:
Don't create random user, just pass user into this class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1256: [core] add negative bucket mode
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #1256:
URL: https://github.com/apache/incubator-paimon/pull/1256#discussion_r1209716156
##########
paimon-core/src/main/java/org/apache/paimon/append/CompactionTask.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.paimon.append;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.io.NewFilesIncrement;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.SerializationUtils.deserializedBytes;
+import static org.apache.paimon.utils.SerializationUtils.serializeBytes;
+
+/**
+ * Compaction task generated by {@link AppendOnlyTableCompactionCoordinator} and executed be {@link
+ * AppendOnlyTableCompactionWorker}.
+ */
+public class CompactionTask implements CommitMessage {
Review Comment:
It does not need to implement `CommitMessage`.
##########
paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java:
##########
@@ -158,4 +159,13 @@ private AppendOnlyCompactManager.CompactRewriter compactRewriter(
return rewriter.result();
};
}
+
+ // overwrite flag in AppendOnlyFileStoreWrite only disable search for restored files
+ public void fromEmptyWriter() {
+ withOverwrite(true);
Review Comment:
We can just modify `AbstractFileStoreWrite.withOverwrite` to `fromEmptyWriter `
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org