support specify database and table for AppendOnlyCompactionTask

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyCompactionTask.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.append;
 
+import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.io.CompactIncrement;
 import org.apache.paimon.io.DataFileMeta;
@@ -39,13 +40,22 @@ public class AppendOnlyCompactionTask {
     private final List<DataFileMeta> compactBefore;
     private final List<DataFileMeta> compactAfter;
 
+    private final Identifier tableIdentifier;
+
     public AppendOnlyCompactionTask(BinaryRow partition, List<DataFileMeta> files) {
+        this(partition, files, Identifier.EMPTY);
+    }
+
+    public AppendOnlyCompactionTask(
+            BinaryRow partition, List<DataFileMeta> files, Identifier identifier) {
+
         Preconditions.checkArgument(
                 files != null && files.size() > 1,
                 "AppendOnlyCompactionTask need more than one file input.");
         this.partition = partition;
         compactBefore = new ArrayList<>(files);
         compactAfter = new ArrayList<>();
+        this.tableIdentifier = identifier;
     }
 
     public BinaryRow partition() {
@@ -72,8 +82,12 @@ public CommitMessage doCompact(AppendOnlyFileStoreWrite write) throws Exception
                 compactIncrement);
     }
 
+    public Identifier tableIdentifier() {
+        return tableIdentifier;
+    }
+
     public int hashCode() {
-        return Objects.hash(partition, compactBefore, compactAfter);
+        return Objects.hash(partition, compactBefore, compactAfter, tableIdentifier);
     }
 
     @Override
@@ -88,7 +102,8 @@ public boolean equals(Object o) {
         AppendOnlyCompactionTask that = (AppendOnlyCompactionTask) o;
         return Objects.equals(partition, that.partition)
                 && Objects.equals(compactBefore, that.compactBefore)
-                && Objects.equals(compactAfter, that.compactAfter);
+                && Objects.equals(compactAfter, that.compactAfter)
+                && Objects.equals(tableIdentifier, that.tableIdentifier);
     }
 
     @Override
@@ -97,7 +112,8 @@ public String toString() {
                 "CompactionTask {"
                         + "partition = %s, "
                         + "compactBefore = %s, "
-                        + "compactAfter = %s}",
-                partition, compactBefore, compactAfter);
+                        + "compactAfter = %s, "
+                        + "tableIdentifier = %s}",
+                partition, compactBefore, compactAfter, tableIdentifier);
     }
 }

paimon-core/src/main/java/org/apache/paimon/catalog/Identifier.java

@@ -19,6 +19,8 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.annotation.Public;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.StringUtils;
 
 import java.io.Serializable;
@@ -33,6 +35,7 @@
  */
 @Public
 public class Identifier implements Serializable {
+    public static final Identifier EMPTY = new Identifier(null, null);
 
     private static final long serialVersionUID = 1L;
 
@@ -109,4 +112,8 @@ public int hashCode() {
     public String toString() {
         return "Identifier{" + "database='" + database + '\'' + ", table='" + table + '\'' + '}';
     }
+
+    public static RowType schema() {
+        return RowType.builder().fields(DataTypes.STRING(), DataTypes.STRING()).build();
+    }
 }

paimon-core/src/main/java/org/apache/paimon/io/IdentifierSerializer.java

@@ -0,0 +1,47 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.utils.ObjectSerializer;
+
+/** Serializer for {@link Identifier}. */
+public class IdentifierSerializer extends ObjectSerializer<Identifier> {
+
+    public IdentifierSerializer() {
+        super(Identifier.schema());
+    }
+
+    @Override
+    public InternalRow toRow(Identifier record) {
+        return GenericRow.of(
+                BinaryString.fromString(record.getDatabaseName()),
+                BinaryString.fromString(record.getObjectName()));
+    }
+
+    @Override
+    public Identifier fromRow(InternalRow rowData) {
+        String databaseName = rowData.isNullAt(0) ? null : rowData.getString(0).toString();
+        String tableName = rowData.isNullAt(1) ? null : rowData.getString(1).toString();
+        return Identifier.create(databaseName, tableName);
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/FileStoreTableFactory.java

@@ -71,6 +71,7 @@ public static FileStoreTable create(FileIO fileIO, Options options) {
     }
 
     public static FileStoreTable create(FileIO fileIO, Path tablePath, TableSchema tableSchema) {
+//        Lock.factory()
         return create(
                 fileIO,
                 tablePath,

paimon-core/src/main/java/org/apache/paimon/table/sink/CompactionTaskSerializer.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.io.DataInputView;
 import org.apache.paimon.io.DataOutputView;
 import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.io.IdentifierSerializer;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -37,12 +38,15 @@
 /** Serializer for {@link AppendOnlyCompactionTask}. */
 public class CompactionTaskSerializer implements VersionedSerializer<AppendOnlyCompactionTask> {
 
-    private static final int CURRENT_VERSION = 2;
+    private static final int CURRENT_VERSION = 3;
 
     private final DataFileMetaSerializer dataFileSerializer;
 
+    private final IdentifierSerializer identifierSerializer;
+
     public CompactionTaskSerializer() {
         this.dataFileSerializer = new DataFileMetaSerializer();
+        this.identifierSerializer = new IdentifierSerializer();
     }
 
     @Override
@@ -69,6 +73,7 @@ public void serializeList(List<AppendOnlyCompactionTask> list, DataOutputView vi
     private void serialize(AppendOnlyCompactionTask task, DataOutputView view) throws IOException {
         serializeBinaryRow(task.partition(), view);
         dataFileSerializer.serializeList(task.compactBefore(), view);
+        identifierSerializer.serialize(task.tableIdentifier(), view);
     }
 
     @Override
@@ -103,6 +108,8 @@ private void checkVersion(int version) {
 
     private AppendOnlyCompactionTask deserialize(DataInputView view) throws IOException {
         return new AppendOnlyCompactionTask(
-                deserializeBinaryRow(view), dataFileSerializer.deserializeList(view));
+                deserializeBinaryRow(view),
+                dataFileSerializer.deserializeList(view),
+                identifierSerializer.deserialize(view));
     }
 }

paimon-core/src/test/java/org/apache/paimon/table/sink/CompactionTaskSerializerTest.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.table.sink;
 
 import org.apache.paimon.append.AppendOnlyCompactionTask;
+import org.apache.paimon.catalog.Identifier;
 
 import org.junit.jupiter.api.Test;
 
@@ -33,12 +34,27 @@ public class CompactionTaskSerializerTest {
 
     @Test
     public void testCompactionTaskSerializer() throws IOException {
-        CompactionTaskSerializer serializer = new CompactionTaskSerializer();
-        AppendOnlyCompactionTask task =
-                new AppendOnlyCompactionTask(row(0), randomNewFilesIncrement().newFiles());
-
-        byte[] bytes = serializer.serialize(task);
-        AppendOnlyCompactionTask task1 = serializer.deserialize(serializer.getVersion(), bytes);
-        assertThat(task).isEqualTo(task1);
+        {
+            CompactionTaskSerializer serializer = new CompactionTaskSerializer();
+            AppendOnlyCompactionTask task =
+                    new AppendOnlyCompactionTask(row(0), randomNewFilesIncrement().newFiles());
+
+            byte[] bytes = serializer.serialize(task);
+            AppendOnlyCompactionTask task1 = serializer.deserialize(serializer.getVersion(), bytes);
+            assertThat(task).isEqualTo(task1);
+        }
+
+        {
+            CompactionTaskSerializer serializer = new CompactionTaskSerializer();
+            AppendOnlyCompactionTask task =
+                    new AppendOnlyCompactionTask(
+                            row(0),
+                            randomNewFilesIncrement().newFiles(),
+                            Identifier.create("db", "table"));
+
+            byte[] bytes = serializer.serialize(task);
+            AppendOnlyCompactionTask task1 = serializer.deserialize(serializer.getVersion(), bytes);
+            assertThat(task).isEqualTo(task1);
+        }
     }
 }