Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Enhancement] support iceberg incremental scan range #52248

Merged
merged 1 commit into from
Oct 25, 2024
Merged

[Enhancement] support iceberg incremental scan range #52248

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/catalog/IcebergTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,7 +303,6 @@ public org.apache.iceberg.Table getNativeTable() {
return nativeTable;
}


@Override
public TTableDescriptor toThrift(List<DescriptorTable.ReferencedPartitionInfo> partitions) {
Preconditions.checkNotNull(partitions);
Expand Down
9 changes: 9 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/connector/CatalogConnectorMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down
9 changes: 9 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/connector/ConnectorMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 {
/**
Expand Down Expand Up @@ -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");
}
}

7 changes: 6 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/connector/GetRemoteFilesParams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down
7 changes: 7 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/connector/RemoteFileInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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{");
Expand Down
Loading
Loading