Skip to content

Make IcebergTableHandle.enforcedPredicate coordinator-only #12766

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

Closed
findepi wants to merge 2 commits into from
Closed

Make IcebergTableHandle.enforcedPredicate coordinator-only #12766

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
40a9a9a
Invoke main IcebergTableHandle constructor
findepi Jun 9, 2022
c3d1bb0
Make IcebergTableHandle.enforcedPredicate coordinator-only
findepi Jun 9, 2022
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
11 changes: 8 additions & 3 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,7 +325,9 @@ public IcebergTableHandle getTableHandle(
table.location(),
table.properties(),
NO_RETRIES,
ImmutableList.of());
ImmutableList.of(),
false,
Optional.empty());
}

@Override
Expand Down Expand Up @@ -1716,7 +1718,8 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(C
}

return Optional.of(new ConstraintApplicationResult<>(
new IcebergTableHandle(table.getSchemaName(),
new IcebergTableHandle(
table.getSchemaName(),
table.getTableName(),
table.getTableType(),
table.getSnapshotId(),
Expand All @@ -1730,7 +1733,9 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(C
table.getTableLocation(),
table.getStorageProperties(),
table.getRetryMode(),
table.getUpdatedColumns()),
table.getUpdatedColumns(),
table.isRecordScannedFiles(),
table.getMaxScannedFileSize()),
remainingConstraint.transformKeys(ColumnHandle.class::cast),
false));
}
Expand Down
56 changes: 48 additions & 8 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTableHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,8 @@ public class IcebergTableHandle
// Filter used during split generation and table scan, but not required to be strictly enforced by Iceberg Connector
private final TupleDomain<IcebergColumnHandle> unenforcedPredicate;

// Filter guaranteed to be enforced by Iceberg connector
private final TupleDomain<IcebergColumnHandle> enforcedPredicate;
// Filter guaranteed to be enforced by Iceberg connector. Coordinator-only
private final Optional<TupleDomain<IcebergColumnHandle>> enforcedPredicate;

private final Set<IcebergColumnHandle> projectedColumns;
private final Optional<String> nameMappingJson;
Expand All @@ -65,7 +65,8 @@ public class IcebergTableHandle
private final Optional<DataSize> maxScannedFileSize;

@JsonCreator
public IcebergTableHandle(
@Deprecated // For JSON deserialization only
public static IcebergTableHandle fromJson(
@JsonProperty("schemaName") String schemaName,
@JsonProperty("tableName") String tableName,
@JsonProperty("tableType") TableType tableType,
Expand All @@ -74,15 +75,14 @@ public IcebergTableHandle(
@JsonProperty("partitionSpecJson") String partitionSpecJson,
@JsonProperty("formatVersion") int formatVersion,
@JsonProperty("unenforcedPredicate") TupleDomain<IcebergColumnHandle> unenforcedPredicate,
@JsonProperty("enforcedPredicate") TupleDomain<IcebergColumnHandle> enforcedPredicate,
@JsonProperty("projectedColumns") Set<IcebergColumnHandle> projectedColumns,
@JsonProperty("nameMappingJson") Optional<String> nameMappingJson,
@JsonProperty("tableLocation") String tableLocation,
@JsonProperty("storageProperties") Map<String, String> storageProperties,
@JsonProperty("retryMode") RetryMode retryMode,
@JsonProperty("updatedColumns") List<IcebergColumnHandle> updatedColumns)
{
this(
return new IcebergTableHandle(
schemaName,
tableName,
tableType,
Expand All @@ -91,7 +91,7 @@ public IcebergTableHandle(
partitionSpecJson,
formatVersion,
unenforcedPredicate,
enforcedPredicate,
Optional.empty(),
projectedColumns,
nameMappingJson,
tableLocation,
Expand Down Expand Up @@ -120,6 +120,45 @@ public IcebergTableHandle(
List<IcebergColumnHandle> updatedColumns,
boolean recordScannedFiles,
Optional<DataSize> maxScannedFileSize)
{
this(
schemaName,
tableName,
tableType,
snapshotId,
tableSchemaJson,
partitionSpecJson,
formatVersion,
unenforcedPredicate,
Optional.of(enforcedPredicate),
projectedColumns,
nameMappingJson,
tableLocation,
storageProperties,
retryMode,
updatedColumns,
recordScannedFiles,
maxScannedFileSize);
}

private IcebergTableHandle(
String schemaName,
String tableName,
TableType tableType,
Optional<Long> snapshotId,
String tableSchemaJson,
String partitionSpecJson,
int formatVersion,
TupleDomain<IcebergColumnHandle> unenforcedPredicate,
Optional<TupleDomain<IcebergColumnHandle>> enforcedPredicate,
Set<IcebergColumnHandle> projectedColumns,
Optional<String> nameMappingJson,
String tableLocation,
Map<String, String> storageProperties,
RetryMode retryMode,
List<IcebergColumnHandle> updatedColumns,
boolean recordScannedFiles,
Optional<DataSize> maxScannedFileSize)
{
this.schemaName = requireNonNull(schemaName, "schemaName is null");
this.tableName = requireNonNull(tableName, "tableName is null");
Expand Down Expand Up @@ -188,10 +227,11 @@ public TupleDomain<IcebergColumnHandle> getUnenforcedPredicate()
return unenforcedPredicate;
}

@JsonProperty
// do not serialize, not needed on workers
@JsonIgnore
public TupleDomain<IcebergColumnHandle> getEnforcedPredicate()
{
return enforcedPredicate;
return enforcedPredicate.orElseThrow(() -> new IllegalStateException("enforcedPredicate not set"));
}

@JsonProperty
Expand Down