Skip to content

[core] Fix FallbackReadScan#plan mixing partition and data filters #7388

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

Open
JunRuiLee wants to merge 1 commit into
base: master
Choose a base branch
from
+177 −51
Open

[core] Fix FallbackReadScan#plan mixing partition and data filters #7388

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
58 changes: 16 additions & 42 deletions paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,11 +130,9 @@ public static class ChainTableBatchScan extends FallbackReadScan {

private final RowDataToObjectArrayConverter partitionConverter;
private final InternalRowPartitionComputer partitionComputer;
private final TableSchema tableSchema;
private final CoreOptions options;
private final RecordComparator partitionComparator;
private final ChainGroupReadTable chainGroupReadTable;
private PartitionPredicate partitionPredicate;
private Predicate dataPredicate;
private Filter<Integer> bucketFilter;

Expand All @@ -143,8 +141,12 @@ public ChainTableBatchScan(
DataTableScan fallbackScan,
TableSchema tableSchema,
ChainGroupReadTable chainGroupReadTable) {
super(mainScan, fallbackScan);
this.tableSchema = tableSchema;
super(
mainScan,
fallbackScan,
chainGroupReadTable.wrapped,
chainGroupReadTable.fallback(),
tableSchema);
this.options = CoreOptions.fromMap(tableSchema.options());
this.chainGroupReadTable = chainGroupReadTable;
this.partitionConverter =
Expand All @@ -169,7 +171,6 @@ public ChainTableBatchScan withFilter(Predicate predicate) {
predicate,
tableSchema.logicalRowType(),
tableSchema.partitionKeys());
setPartitionPredicate(pair.getLeft().orElse(null));
dataPredicate =
pair.getRight().isEmpty() ? null : PredicateBuilder.and(pair.getRight());
}
Expand All @@ -179,57 +180,30 @@ public ChainTableBatchScan withFilter(Predicate predicate) {
@Override
public ChainTableBatchScan withPartitionFilter(Map<String, String> partitionSpec) {
super.withPartitionFilter(partitionSpec);
if (partitionSpec != null) {
setPartitionPredicate(
PartitionPredicate.fromMap(
tableSchema.logicalPartitionType(),
partitionSpec,
options.partitionDefaultName()));
}
return this;
}

@Override
public ChainTableBatchScan withPartitionFilter(List<BinaryRow> partitions) {
super.withPartitionFilter(partitions);
if (partitions != null) {
setPartitionPredicate(
PartitionPredicate.fromMultiple(
tableSchema.logicalPartitionType(), partitions));
}
return this;
}

@Override
public ChainTableBatchScan withPartitionsFilter(List<Map<String, String>> partitions) {
super.withPartitionsFilter(partitions);
if (partitions != null) {
setPartitionPredicate(
PartitionPredicate.fromMaps(
tableSchema.logicalPartitionType(),
partitions,
options.partitionDefaultName()));
}
return this;
}

@Override
public ChainTableBatchScan withPartitionFilter(PartitionPredicate partitionPredicate) {
super.withPartitionFilter(partitionPredicate);
if (partitionPredicate != null) {
setPartitionPredicate(partitionPredicate);
}
return this;
}

@Override
public ChainTableBatchScan withPartitionFilter(Predicate partitionPredicate) {
super.withPartitionFilter(partitionPredicate);
if (partitionPredicate != null) {
setPartitionPredicate(
PartitionPredicate.fromPredicate(
tableSchema.logicalPartitionType(), partitionPredicate));
}
return this;
}

Expand All @@ -252,6 +226,7 @@ public ChainTableBatchScan withBucketFilter(Filter<Integer> bucketFilter) {
@Override
public Plan plan() {
List<Split> splits = new ArrayList<>();
PartitionPredicate partitionPredicate = getPartitionPredicate();
PredicateBuilder builder = new PredicateBuilder(tableSchema.logicalPartitionType());
for (Split split : mainScan.plan().splits()) {
DataSplit dataSplit = (DataSplit) split;
Expand All @@ -271,9 +246,11 @@ public Plan plan() {

Set<BinaryRow> snapshotPartitions =
new HashSet<>(
newPartitionListingScan(true, partitionPredicate).listPartitions());
newChainPartitionListingScan(true, partitionPredicate)
.listPartitions());

DataTableScan deltaPartitionScan = newPartitionListingScan(false, partitionPredicate);
DataTableScan deltaPartitionScan =
newChainPartitionListingScan(false, partitionPredicate);
List<BinaryRow> deltaPartitions =
deltaPartitionScan.listPartitions().stream()
.filter(p -> !snapshotPartitions.contains(p))
Expand All @@ -292,7 +269,7 @@ public Plan plan() {
PartitionPredicate.fromPredicate(
tableSchema.logicalPartitionType(), snapshotPredicate);
DataTableScan snapshotPartitionsScan =
newPartitionListingScan(true, snapshotPartitionPredicate);
newChainPartitionListingScan(true, snapshotPartitionPredicate);
List<BinaryRow> candidateSnapshotPartitions =
snapshotPartitionsScan.listPartitions();
candidateSnapshotPartitions =
Expand Down Expand Up @@ -393,8 +370,9 @@ public Plan plan() {

@Override
public List<PartitionEntry> listPartitionEntries() {
DataTableScan snapshotScan = newPartitionListingScan(true, partitionPredicate);
DataTableScan deltaScan = newPartitionListingScan(false, partitionPredicate);
PartitionPredicate partitionPredicate = getPartitionPredicate();
DataTableScan snapshotScan = newChainPartitionListingScan(true, partitionPredicate);
DataTableScan deltaScan = newChainPartitionListingScan(false, partitionPredicate);
List<PartitionEntry> partitionEntries =
new ArrayList<>(snapshotScan.listPartitionEntries());
Set<BinaryRow> partitions =
Expand All @@ -408,11 +386,7 @@ public List<PartitionEntry> listPartitionEntries() {
return partitionEntries;
}

private void setPartitionPredicate(PartitionPredicate predicate) {
this.partitionPredicate = predicate;
}

private DataTableScan newPartitionListingScan(
private DataTableScan newChainPartitionListingScan(
boolean snapshot, PartitionPredicate scanPartitionPredicate) {
DataTableScan scan =
snapshot
Expand Down
100 changes: 92 additions & 8 deletions paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SegmentsCache;

Expand Down Expand Up @@ -189,7 +190,8 @@ protected Map<String, String> rewriteFallbackOptions(Map<String, String> options
@Override
public DataTableScan newScan() {
validateSchema();
return new FallbackReadScan(wrapped.newScan(), fallback.newScan());
return new FallbackReadScan(
wrapped.newScan(), fallback.newScan(), wrapped, fallback, wrapped.schema());
}

protected void validateSchema() {
Expand Down Expand Up @@ -356,10 +358,22 @@ public static class FallbackReadScan implements DataTableScan {

protected final DataTableScan mainScan;
protected final DataTableScan fallbackScan;

public FallbackReadScan(DataTableScan mainScan, DataTableScan fallbackScan) {
protected final FileStoreTable wrappedTable;
protected final FileStoreTable fallbackTable;
protected final TableSchema tableSchema;
private PartitionPredicate partitionPredicate;

public FallbackReadScan(
DataTableScan mainScan,
DataTableScan fallbackScan,
FileStoreTable wrappedTable,
FileStoreTable fallbackTable,
TableSchema tableSchema) {
this.mainScan = mainScan;
this.fallbackScan = fallbackScan;
this.wrappedTable = wrappedTable;
this.fallbackTable = fallbackTable;
this.tableSchema = tableSchema;
}

@Override
Expand All @@ -373,6 +387,14 @@ public FallbackReadScan withShard(int indexOfThisSubtask, int numberOfParallelSu
public FallbackReadScan withFilter(Predicate predicate) {
mainScan.withFilter(predicate);
fallbackScan.withFilter(predicate);
if (predicate != null) {
Pair<Optional<PartitionPredicate>, List<Predicate>> pair =
PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
predicate,
tableSchema.logicalRowType(),
tableSchema.partitionKeys());
setPartitionPredicate(pair.getLeft().orElse(null));
}
return this;
}

Expand All @@ -387,27 +409,61 @@ public FallbackReadScan withLimit(int limit) {
public FallbackReadScan withPartitionFilter(Map<String, String> partitionSpec) {
mainScan.withPartitionFilter(partitionSpec);
fallbackScan.withPartitionFilter(partitionSpec);
if (partitionSpec != null) {
setPartitionPredicate(
PartitionPredicate.fromMap(
tableSchema.logicalPartitionType(),
partitionSpec,
CoreOptions.fromMap(tableSchema.options()).partitionDefaultName()));
}
return this;
}

@Override
public FallbackReadScan withPartitionFilter(List<BinaryRow> partitions) {
mainScan.withPartitionFilter(partitions);
fallbackScan.withPartitionFilter(partitions);
if (partitions != null) {
setPartitionPredicate(
PartitionPredicate.fromMultiple(
tableSchema.logicalPartitionType(), partitions));
}
return this;
}

@Override
public InnerTableScan withPartitionsFilter(List<Map<String, String>> partitions) {
mainScan.withPartitionsFilter(partitions);
fallbackScan.withPartitionsFilter(partitions);
if (partitions != null) {
setPartitionPredicate(
PartitionPredicate.fromMaps(
tableSchema.logicalPartitionType(),
partitions,
CoreOptions.fromMap(tableSchema.options()).partitionDefaultName()));
}
return this;
}

@Override
public InnerTableScan withPartitionFilter(PartitionPredicate partitionPredicate) {
mainScan.withPartitionFilter(partitionPredicate);
fallbackScan.withPartitionFilter(partitionPredicate);
if (partitionPredicate != null) {
setPartitionPredicate(partitionPredicate);
}
return this;
}

@Override
public FallbackReadScan withPartitionFilter(Predicate partitionPredicate) {
mainScan.withPartitionFilter(partitionPredicate);
fallbackScan.withPartitionFilter(partitionPredicate);
if (partitionPredicate != null) {
setPartitionPredicate(
PartitionPredicate.fromPredicate(
tableSchema.logicalPartitionType(), partitionPredicate));
}
return this;
}

Expand Down Expand Up @@ -446,18 +502,26 @@ public InnerTableScan dropStats() {
return this;
}

/**
* Builds a plan for fallback read.
*
* <p>Partitions that exist in the main branch (based on partition predicates only) are
* treated as complete and are read from the main branch with the full predicate. Partitions
* that exist only in the fallback branch are read from the fallback branch.
*/
@Override
public TableScan.Plan plan() {
List<Split> splits = new ArrayList<>();
Set<BinaryRow> completePartitions = new HashSet<>();
Set<BinaryRow> completePartitions =
new HashSet<>(
newPartitionListingScan(true, partitionPredicate).listPartitions());
for (Split split : mainScan.plan().splits()) {
DataSplit dataSplit = (DataSplit) split;
splits.add(toFallbackSplit(dataSplit, false));
completePartitions.add(dataSplit.partition());
}

List<BinaryRow> remainingPartitions =
fallbackScan.listPartitions().stream()
newPartitionListingScan(false, partitionPredicate).listPartitions().stream()
.filter(p -> !completePartitions.contains(p))
.collect(Collectors.toList());
if (!remainingPartitions.isEmpty()) {
Expand All @@ -471,18 +535,38 @@ public TableScan.Plan plan() {

@Override
public List<PartitionEntry> listPartitionEntries() {
DataTableScan mainListingScan = newPartitionListingScan(true, partitionPredicate);
DataTableScan fallbackListingScan = newPartitionListingScan(false, partitionPredicate);
List<PartitionEntry> partitionEntries =
new ArrayList<>(mainScan.listPartitionEntries());
new ArrayList<>(mainListingScan.listPartitionEntries());
Set<BinaryRow> partitions =
partitionEntries.stream()
.map(PartitionEntry::partition)
.collect(Collectors.toSet());
List<PartitionEntry> fallBackPartitionEntries = fallbackScan.listPartitionEntries();
List<PartitionEntry> fallBackPartitionEntries =
fallbackListingScan.listPartitionEntries();
fallBackPartitionEntries.stream()
.filter(e -> !partitions.contains(e.partition()))
.forEach(partitionEntries::add);
return partitionEntries;
}

protected void setPartitionPredicate(PartitionPredicate predicate) {
this.partitionPredicate = predicate;
}

protected PartitionPredicate getPartitionPredicate() {
return partitionPredicate;
}

private DataTableScan newPartitionListingScan(
boolean isMain, PartitionPredicate scanPartitionPredicate) {
DataTableScan scan = isMain ? wrappedTable.newScan() : fallbackTable.newScan();
if (scanPartitionPredicate != null) {
scan.withPartitionFilter(scanPartitionPredicate);
}
return scan;
}
}

@Override
Expand Down
7 changes: 6 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,12 @@ private SnapshotReader newSnapshotReader(FileStoreTable wrapped) {
public DataTableScan newScan() {
if (wrapped instanceof FallbackReadFileStoreTable) {
FallbackReadFileStoreTable table = (FallbackReadFileStoreTable) wrapped;
return new FallbackReadScan(newScan(table.wrapped()), newScan(table.fallback()));
return new FallbackReadScan(
newScan(table.wrapped()),
newScan(table.fallback()),
table.wrapped(),
table.fallback(),
table.wrapped().schema());
}
return newScan(wrapped);
}
Expand Down
Loading