apache / apache/fluss

Limit Scan will emit wrong data for partitioned table in partition filter scan

Open
#2,353 1 comment 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
2.1k
Forks
625
Avg merge
3d 14h
Merged PRs (30d)
97

Description

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Fluss version

0.8.0 (latest release)

### Please describe the bug 🐞

Can be reproduce by the following test, add a new test method in FlinkUnionReadPrimaryKeyTableITCase
```
@Test
void t1() throws Exception {
boolean isPartitioned = true;
JobClient jobClient = buildTieringJob(execEnv);

String tableName = "pk_table_full" + (isPartitioned ? "_partitioned" : "_non_partitioned");
TablePath t1 = TablePath.of(DEFAULT_DB, tableName);
Map bucketLogEndOffset = new HashMap<>();
// create table & write initial data
long tableId =
preparePKTableFullType(t1, DEFAULT_BUCKET_NUM, isPartitioned, bucketLogEndOffset);

// // wait unit records have been synced
// waitUntilBucketSynced(t1, tableId, DEFAULT_BUCKET_NUM, isPartitioned);
//
// // check the status of replica after synced
// assertReplicaStatus(t1, tableId, DEFAULT_BUCKET_NUM, isPartitioned, bucketLogEndOffset);

// will read paimon snapshot, won't merge log since it's empty
List resultEmptyLog =
toSortedRows(
batchTEnv.executeSql(
"select * from " + tableName + " where c16 = 'not_exists' limit 10"));
System.out.println(resultEmptyLog);

jobClient.cancel().get();
}
```

Then print
```
[+I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182Z, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], not_exists], +I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182Z, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], not_exists], +I[true, 10, 20, 30, 40, 50.1, 60.0, another_string, 0.90, 100, 2023-10-25T12:01:13.200Z, 2023-10-25T12:01:13.200005Z, 2023-10-25T12:01:13.201, 2023-10-25T12:01:13.201006, [1, 2, 3, 4], not_exists], +I[true, 10, 20, 30, 40, 50.1, 60.0, another_string, 0.90, 100, 2023-10-25T12:01:13.200Z, 2023-10-25T12:01:13.200005Z, 2023-10-25T12:01:13.201, 2023-10-25T12:01:13.201006, [1, 2, 3, 4], not_exists]]

```

Note the main branch won't have the issue, the reason is after #1934 , the filter won't be accept, so the limit won't push down.

But I create the issue to track it in case some day that we accept the filter again, the reason is that the source accept the partition filter, and then do the limit scan, but the limit scan don't respect the partition filter and return the result. Flink'll append the partition predicate `'not_exists'` into the scan row, which cause wrong result.

### Solution

_No response_

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with FlinkUnionReadPrimaryKeyTableITCase and add the reproduction described in the issue for a partitioned table with c16 = 'not_exists'. Read the partition-filter and limit-scan path, including the change referenced in #1934, then verify that the scan respects the partition filter and does not return rows with an appended partition value.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.