apache / apache/paimon

[Bug] LookupStreamingReader should not apply valueFilter in scanning kv table

Open
#3,605 0 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
3.4k
Forks
1.4k
Avg merge
1d 11h
Merged PRs (30d)
396

Description

### Search before asking

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

### Paimon version

0.8.0

### Compute Engine

flink 1.17.1

### Minimal reproduce step

```

CREATE TABLE paimon.xxx.dim_0625 (
id bigint,
k string,
v string,
PRIMARY KEY (id) NOT ENFORCED
)
WITH (
'bucket' = '1'
);

CREATE TABLE paimon.xxx.fact_0625 (
name string,
k string,
proctime as PROCTIME()
)
WITH (
'bucket' = '1'
);

select
a.name,
a.k as ak,
b.k as bk,
b.v
from paimon.xxx.fact_0625 /*+ OPTIONS('scan.mode'='latest') */ a
left join paimon.xxx.dim_0625 /*+ OPTIONS('continuous.discovery-interval'='3s') */ FOR SYSTEM_TIME AS OF a.proctime AS b
on a.k = b.k
where b.v = 'x'
```

Run the sqls above in streaming mode, the operator in DAG would be, Note that it's turned into InnerJoin by flink because of the where condiction on lookup table:
```
[5690]:TableSourceScan(table=[[paimon, xxx, fact_0625]], fields=[name, k], hints=[[[OPTIONS options:{scan.mode=latest}]]])
+- [5691]:LookupJoin(table=[paimon.xxx.dim_0625], joinType=[InnerJoin], lookup=[v=_UTF-16LE'x', k=k], where=[(v = 'x')], select=[name, k, k])
+- [5692]:Calc(select=[name, k AS ak, k0 AS bk, CAST('x' AS VARCHAR(2147483647)) AS v])
+- [5693]:Sink(table=[default_cat.default_db.print], fields=[name, ak, bk, v])
```

Now run there sqls in batch mode sequentially to produce records in both dim and fact tables:
```
insert into paimon.xxx.dim_0625
select 1 as id, 'k' as k, 'x' as v;

insert into paimon.xxx.fact_0625
select 'r1' as id, 'k' as k;

insert into paimon.xxx.dim_0625
select 1 as id, 'k' as k, 'y' as v;

(now dim table should contains no records with 'x' in field v)

insert into paimon.xxx.fact_0625
select 'r2' as id, 'k' as k;

insert into paimon.xxx.fact_0625
select 'r3' as id, 'k' as k;

```

The output will be :
```
+I[r1, k, k, x]
+I[r2, k, k, x]
+I[r3, k, k, x]
```

### What doesn't meet your expectations?

+I[r2, k, k, x]
+I[r3, k, k, x]

Those two records should not join out with x.

### Anything else?

In this case the predicate [v = 'x'] will be applied to Scan used by LookupStreamingReader,
![image](https://github.com/apache/paimon/assets/14367839/3be8f53f-4f15-4a2d-b4bc-e71bc440d78d)
And the predicate will finally become valueFilter in KeyValueFileStoreScan, and used in filterWholeBucketByStats,
causing the v='y' record (and the whole lsm tree) being filtered out by the stats of the ManifestEntry.

But on the other hand, the old version of the record in lookup table is waiting for a UPDATE_AFTER record with the same key to retract itself. (See SecondaryIndexLookupTable.refresh() )
![image](https://github.com/apache/paimon/assets/14367839/cc9dfa9d-f2b4-401c-8940-47020e3004d2)

### 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 by tracing LookupStreamingReader and its use of KeyValueFileStoreScan, then inspect how valueFilter reaches filterWholeBucketByStats. Review SecondaryIndexLookupTable.refresh() and reproduce the issue with the SQL steps in the report; done means lookup updates do not emit stale x records after the dimension value changes to y.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, sql
Domain
databases, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.