iceberg async

Signed-off-by: stephen <[email protected]>
StarRocks · Oct 24, 2024 · d865820 · d865820
1 parent 1ed16a9
commit d865820
Show file tree

Hide file tree

Showing 34 changed files with 1,801 additions and 971 deletions.
diff --git a/fe/fe-core/src/main/java/com/starrocks/catalog/IcebergTable.java b/fe/fe-core/src/main/java/com/starrocks/catalog/IcebergTable.java
@@ -303,7 +303,6 @@ public org.apache.iceberg.Table getNativeTable() {
         return nativeTable;
     }
 
-
     @Override
     public TTableDescriptor toThrift(List<DescriptorTable.ReferencedPartitionInfo> partitions) {
         Preconditions.checkNotNull(partitions);

diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/CatalogConnectorMetadata.java b/fe/fe-core/src/main/java/com/starrocks/connector/CatalogConnectorMetadata.java
@@ -17,6 +17,7 @@
 import com.google.common.collect.ImmutableList;
 import com.starrocks.catalog.Column;
 import com.starrocks.catalog.Database;
+import com.starrocks.catalog.IcebergTable;
 import com.starrocks.catalog.MaterializedIndexMeta;
 import com.starrocks.catalog.PartitionKey;
 import com.starrocks.catalog.Table;
@@ -56,10 +57,13 @@
 import com.starrocks.sql.optimizer.operator.scalar.ScalarOperator;
 import com.starrocks.sql.optimizer.statistics.Statistics;
 import com.starrocks.thrift.TSinkCommitInfo;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileContent;
 
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.starrocks.catalog.system.information.InfoSchemaDb.isInfoSchemaDb;
@@ -211,6 +215,11 @@ public List<PartitionKey> getPrunedPartitions(Table table, ScalarOperator predic
         return normal.getPrunedPartitions(table, predicate, limit, version);
     }
 
+    @Override
+    public Set<DeleteFile> getDeleteFiles(IcebergTable table, Long snapshotId, ScalarOperator predicate, FileContent content) {
+        return normal.getDeleteFiles(table, snapshotId, predicate, content);
+    }
+
     @Override
     public void clear() {
         normal.clear();

diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/ConnectorMetadata.java b/fe/fe-core/src/main/java/com/starrocks/connector/ConnectorMetadata.java
@@ -17,6 +17,7 @@
 import com.google.common.collect.Lists;
 import com.starrocks.catalog.Column;
 import com.starrocks.catalog.Database;
+import com.starrocks.catalog.IcebergTable;
 import com.starrocks.catalog.MaterializedIndexMeta;
 import com.starrocks.catalog.PartitionKey;
 import com.starrocks.catalog.Table;
@@ -54,12 +55,15 @@
 import com.starrocks.sql.optimizer.operator.scalar.ScalarOperator;
 import com.starrocks.sql.optimizer.statistics.Statistics;
 import com.starrocks.thrift.TSinkCommitInfo;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileContent;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 
 public interface ConnectorMetadata {
     /**
@@ -339,5 +343,10 @@ default void alterView(AlterViewStmt stmt) throws DdlException, UserException {
     default CloudConfiguration getCloudConfiguration() {
         throw new StarRocksConnectorException("This connector doesn't support getting cloud configuration");
     }
+
+    default Set<DeleteFile> getDeleteFiles(IcebergTable icebergTable, Long snapshotId,
+                                           ScalarOperator predicate, FileContent fileContent) {
+        throw new StarRocksConnectorException("This connector doesn't support getting delete files");
+    }
 }
 
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/GetRemoteFilesParams.java b/fe/fe-core/src/main/java/com/starrocks/connector/GetRemoteFilesParams.java
@@ -30,7 +30,7 @@ public class GetRemoteFilesParams {
     private boolean useCache = true;
     private boolean checkPartitionExistence = true;
 
-    private GetRemoteFilesParams(Builder builder) {
+    protected GetRemoteFilesParams(Builder builder) {
         this.partitionKeys = builder.partitionKeys;
         this.partitionNames = builder.partitionNames;
         this.partitionAttachments = builder.partitionAttachments;
@@ -56,6 +56,11 @@ public GetRemoteFilesParams copy() {
                 .build();
     }
 
+    @SuppressWarnings("unchecked")
+    public  <T extends GetRemoteFilesParams> T cast() {
+        return (T) this;
+    }
+
     // Getters
     public List<PartitionKey> getPartitionKeys() {
         return partitionKeys;

diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/RemoteFileInfo.java b/fe/fe-core/src/main/java/com/starrocks/connector/RemoteFileInfo.java
@@ -20,6 +20,8 @@
 import java.util.List;
 
 public class RemoteFileInfo {
+    public static RemoteFileInfo EMPTY = new RemoteFileInfo();
+
     private RemoteFileInputFormat format;
     private List<RemoteFileDesc> files;
     private String fullPath;
@@ -64,6 +66,11 @@ public void setAttachment(Object attachment) {
         this.attachment = attachment;
     }
 
+    @SuppressWarnings("unchecked")
+    public  <T extends RemoteFileInfo> T cast() {
+        return (T) this;
+    }
+
     @Override
     public String toString() {
         final StringBuilder sb = new StringBuilder("RemoteFileInfo{");