[parquet] Merge the file index and the deletion vector and push down filtering

[Tan-JiaLiang]

Purpose

Currently, the bitmap index result has been supported pushdown in Parquet, We can merge the deletion vector and push down filtering.

Tests

org.apache.paimon.table.PrimaryKeyFileStoreTableTest#testDeletionVectorsCombineWithFileIndexPushDownParquet

API and Format

Documentation

[JingsongLi]

Can you share some benchmark on this? I'm a little worried about whether this optimization is useful for DV and whether it has any side effects.

[Tan-JiaLiang]

@JingsongLi Sorry late. Your worried is reasonable, when DV result filters nothing, it incurs some performance overhead (the bitmap operation), maybe index have the same problem, because I only test the EQ predicate currently.

This is the benchmark code: https://github.com/Tan-JiaLiang/paimon/tree/benchmark/dv_push_down

I test in a 20 fields table, generate 1000000 rows per commit, commit 10 times, using the BSI index and DV to test the EQ predicate.

• normal: dv and index disable push down.
• dv-push-down: dv enable push down, and disable the index.
• index-push-down: dv disable push down, and index enable.
• dv-and-index-push-down: dv and index enable push down.

generate random value bound in [0, 10000000) for Primary key field, [0, 50000) for BSI index field, using a random value 27024 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-50000-27024	18105 / 18197	165.7	6034.9	1.0X	0	0
dv-push-down-10000000-50000-27024	18577 / 18744	161.5	6192.4	1.0X	0	0
index-push-down-10000000-50000-27024	5272 / 5440	569.0	1757.3	3.4X	0	246
dv-and-index-push-down-10000000-50000-27024	5024 / 5052	597.1	1674.8	3.6X	0	260

generate random value bound in [0, 10000000) for Primary key field, [0, 100000) for BSI index field, using a random value 7748 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-100000-7748	15763 / 15845	190.3	5254.5	1.0X	0	0
dv-push-down-10000000-100000-7748	16123 / 16160	186.1	5374.2	1.0X	0	0
index-push-down-10000000-100000-7748	2413 / 2443	1243.3	804.3	6.5X	0	310
dv-and-index-push-down-10000000-100000-7748	2163 / 2190	1386.9	721.0	7.3X	0	318

generate random value bound in [0, 10000000) for Primary key field, [0, 300000) for BSI index field, using a random value 288225 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-300000-288225	15781 / 15836	190.1	5260.3	1.0X	0	0
dv-push-down-10000000-300000-288225	16102 / 16199	186.3	5367.5	1.0X	0	0
index-push-down-10000000-300000-288225	1623 / 1627	1848.8	540.9	9.7X	0	329
dv-and-index-push-down-10000000-300000-288225	1418 / 1484	2116.3	472.5	11.1X	0	334

generate random value bound in [0, 10000000) for Primary key field, [0, 800000) for BSI index field, using a random value 788897 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-800000-788897	15685 / 15760	191.3	5228.3	1.0X	0	0
dv-push-down-10000000-800000-788897	16049 / 16413	186.9	5349.8	1.0X	0	0
index-push-down-10000000-800000-788897	632 / 676	4744.8	210.8	24.8X	0	263
dv-and-index-push-down-10000000-800000-788897	679 / 696	4420.2	226.2	23.1X	0	263

[Tan-JiaLiang]

Maybe add RoaringBitmap#contains(long, long) to do fast skip is better, I will benchmark it again.

[Tan-JiaLiang]

Hi @JingsongLi ,I found some bugs in RoaringBitmap 1.0.5 version when I'm benchmark. Will you consider upgrade the roaringbitmap version?

• Fix RunContainer#contains(BitmapContainer) (#721) RoaringBitmap/RoaringBitmap#722
• Fix ArrayContainer#contains(RunContainer) (#723) RoaringBitmap/RoaringBitmap#724
• Fix previousValue not returning correct results in some cases RoaringBitmap/RoaringBitmap#736
• Skip over empty words in RunContainer#contains(BitmapContainer) RoaringBitmap/RoaringBitmap#739

[Tan-JiaLiang]

I used the same data as above for testing after upgrade the RoaringBitmap version to 1.2.1.

generate random value bound in [0, 10000000) for Primary key field, [0, 50000) for BSI index field, using a random value 27024 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-50000-27024	16717 / 17656	179.5	5572.4	1.0X	0	0
dv-push-down-10000000-50000-27024	17083 / 17500	175.6	5694.2	1.0X	0	0
index-push-down-10000000-50000-27024	4623 / 4771	649.0	1540.9	3.6X	0	246
dv-and-index-push-down-10000000-50000-27024	4227 / 4353	709.8	1408.9	4.0X	0	260

generate random value bound in [0, 10000000) for Primary key field, [0, 100000) for BSI index field, using a random value 7748 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-100000-7748	16024 / 16530	187.2	5341.3	1.0X	0	0
dv-push-down-10000000-100000-7748	15695 / 16090	191.1	5231.7	1.0X	0	0
index-push-down-10000000-100000-7748	2958 / 3298	1014.1	986.1	5.4X	0	310
dv-and-index-push-down-10000000-100000-7748	2738 / 2756	1095.8	912.6	5.9X	0	318

generate random value bound in [0, 10000000) for Primary key field, [0, 300000) for BSI index field, using a random value 288225 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-300000-288225	16367 / 16460	183.3	5455.8	1.0X	0	0
dv-push-down-10000000-300000-288225	16547 / 16852	181.3	5515.7	1.0X	0	0
index-push-down-10000000-300000-288225	2068 / 2092	1450.7	689.3	7.9X	0	329
dv-and-index-push-down-10000000-300000-288225	1726 / 1773	1737.9	575.4	9.5X	0	334

generate random value bound in [0, 10000000) for Primary key field, [0, 800000) for BSI index field, using a random value 788897 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-800000-788897	16168 / 16287	185.6	5389.3	1.0X	0	0
dv-push-down-10000000-800000-788897	16020 / 16591	187.3	5340.0	1.0X	0	0
index-push-down-10000000-800000-788897	819 / 857	3662.2	273.1	19.7X	0	263
dv-and-index-push-down-10000000-800000-788897	788 / 1123	3806.2	262.7	20.5X	0	263

[JingsongLi]

Hi @JingsongLi ,I found some bugs in RoaringBitmap 1.0.5 version when I'm benchmark. Will you consider upgrade the roaringbitmap version?

It is OK to upgrade it.

[Tan-JiaLiang]

I did another benchmark based on the code submitted so far. I came to some conclusions: (Please correct me if something wrong)

1. DV push down itself is hard to filtering some row ranges because it is sparse and level compaction may do when writing.
2. Based on the current benchmark, support the DV push down should not cause much performance penalty.
3. DV push down combine with index push down can give better performance.

generate random value bound in [0, 10000000) for Primary key field, [0, 50000) for BSI index field, using a random value 27024 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-50000-27024	15584 / 15684	192.5	5194.7	1.0X	0	0
dv-push-down-10000000-50000-27024	15577 / 15746	192.6	5192.3	1.0X	0	0
index-push-down-10000000-50000-27024	4662 / 4745	643.5	1554.0	3.3X	0	246
dv-and-index-push-down-10000000-50000-27024	4088 / 4129	733.9	1362.7	3.8X	0	260

generate random value bound in [0, 10000000) for Primary key field, [0, 100000) for BSI index field, using a random value 7748 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-100000-7748	16040 / 16864	187.0	5346.7	1.0X	0	0
dv-push-down-10000000-100000-7748	16389 / 16792	183.1	5462.9	1.0X	0	0
index-push-down-10000000-100000-7748	2564 / 2616	1170.0	854.7	6.3X	0	310
dv-and-index-push-down-10000000-100000-7748	2234 / 2281	1343.0	744.6	7.2X	0	318

generate random value bound in [0, 10000000) for Primary key field, [0, 300000) for BSI index field, using a random value 288225 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-300000-288225	15347 / 15494	195.5	5115.8	1.0X	0	0
dv-push-down-10000000-300000-288225	15265 / 15435	196.5	5088.2	1.0X	0	0
index-push-down-10000000-300000-288225	1681 / 1702	1784.3	560.5	9.1X	0	329
dv-and-index-push-down-10000000-300000-288225	1533 / 1551	1956.4	511.1	10.0X	0	334

generate random value bound in [0, 10000000) for Primary key field, [0, 800000) for BSI index field, using a random value 788897 to test EQ predicate.

read	Best/Avg Time(ms)	Row Rate(K/s)	Per Row(ns)	Relative	filter row groups	filter row ranges
normal-10000000-800000-788897	15562 / 15574	192.8	5187.2	1.0X	0	0
dv-push-down-10000000-800000-788897	15309 / 15542	196.0	5103.0	1.0X	0	0
index-push-down-10000000-800000-788897	745 / 758	4026.9	248.3	20.9X	0	263
dv-and-index-push-down-10000000-800000-788897	742 / 750	4043.2	247.3	21.0X	0	263

[ranxianglei]

Hi,can you help find which file the benchmark code: https://github.com/Tan-JiaLiang/paimon/tree/benchmark/dv_push_down ? I have review that results, but maybe slow than expexct ,could you check this issue #4586 ?

chinese: 我不知道你是不是中国人，会不会中文，但是我英文非常差，打上面英文字怕你看不懂，特别中文描述下。
是否有benchmark具体的代码位置？在#4586 中，修复过一些问题已经把查询性能提升了很多了，我理解在1千万行数据查询主键，即时查询1000次也仅仅花费4s左右，平均每一次都是10ms以下的。你给你这个测试和我们之前测试的结果差太大了，可以给下测试代码吗？我帮你看下

@Tan-JiaLiang

[Tan-JiaLiang]

Hi, @ranxianglei I'm glad you're willing to help.

You can find the benchmark code here and If something wrong, please feel free to ask.
https://github.com/Tan-JiaLiang/paimon/blob/benchmark/dv_push_down/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/deletionvectors/DeletionVectorsWithIndexResultPushDownBenchmark.java

Actually I did not test the EQ predicate with the primary key, I was testing the EQ predicate with another field, test benchmark with or without index and deletion vector push down.

In order to minimise errors, the benchmark will run several iterations, and every single iteration will run query in many times, so you can see that it takes a long time.

[ranxianglei]

I understand that if the filter condition is pushed down to orc or Parquet, good performance can be obtained without using index push down. It happens that the dv table can do push-down of non-primary keys. I tested Orc's dv table pushdown. The size of the data generated was similar to yours. On average, a non-primary key query took about 100 ms. I tested it locally and the hard disk was SSD. Although it is much slower than the gradual query which takes 10ms, the performance is not bad.

我理解如果把filter条件下推到orc或者Parquet，不需要使用index的下推也可以获得很好的性能。恰巧dv表可以做非主键的下推。我测试了orc的dv表下推，生成的数据规模和你的类似，平均一个非主键查询的耗时大约100多ms，本机测试，硬盘是ssd。虽然比逐渐查询耗时10ms慢很多，但是性能还算可以。当然性能还可以优化，这个case应该也可以做到10ms左右。

如果你想复现需要暂时把 && !deletionVectorsEnabled
&& !(fileIndexResult instanceof BitmapIndexResult 删去，因为这里一个bug尚未修复，但是不影响测试性能。

paimon/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java

Line 281 in ffb1a71

if (!conjunctPredicates.isEmpty()

The reason why this result occurs is that before non-primary key pushdown is enabled in the dv table, the filter condition cannot be applied to the orc index. After the index takes effect, the number of filtered row groups is naturally reduced. I don’t know much about Parquet. Maybe there is still something wrong with Parquet implementation. You can try it yourself.

之所以会出现这个结果是因为dv表在未开启非主键下推之前，filter条件无法作用到orc的索引上，索引起作用后，筛选后的row group数量自然少了。所以实际上不需要用index下推，仅仅用orc或者Parquet自带的稀疏索引或者bloom过滤器可以获得很好的性能，这里要优化的点或许是每次扫描group 元数据过多的问题。

Parquet我不太懂，或许Parquet实现还是有问题，你可以自己试试。

schema

{
    "version": 2,
    "id": 0,
    "fields": [
        {
            "id": 0,
            "name": "f0",
            "type": "INT NOT NULL"
        },
        {
            "id": 1,
            "name": "f1",
            "type": "INT NOT NULL"
        },
        {
            "id": 2,
            "name": "f2",
            "type": "STRING"
        },
        {
            "id": 3,
            "name": "f3",
            "type": "STRING"
        },
        {
            "id": 4,
            "name": "f4",
            "type": "STRING"
        },
        {
            "id": 5,
            "name": "f5",
            "type": "STRING"
        },
        {
            "id": 6,
            "name": "f6",
            "type": "STRING"
        },
        {
            "id": 7,
            "name": "f7",
            "type": "STRING"
        },
        {
            "id": 8,
            "name": "f8",
            "type": "STRING"
        },
        {
            "id": 9,
            "name": "f9",
            "type": "STRING"
        },
        {
            "id": 10,
            "name": "f10",
            "type": "STRING"
        },
        {
            "id": 11,
            "name": "f11",
            "type": "STRING"
        },
        {
            "id": 12,
            "name": "f12",
            "type": "STRING"
        },
        {
            "id": 13,
            "name": "f13",
            "type": "STRING"
        },
        {
            "id": 14,
            "name": "f14",
            "type": "STRING"
        },
        {
            "id": 15,
            "name": "f15",
            "type": "STRING"
        },
        {
            "id": 16,
            "name": "f16",
            "type": "STRING"
        },
        {
            "id": 17,
            "name": "f17",
            "type": "STRING"
        },
        {
            "id": 18,
            "name": "f18",
            "type": "STRING"
        },
        {
            "id": 19,
            "name": "f19",
            "type": "STRING"
        },
        {
            "id": 20,
            "name": "f20",
            "type": "STRING"
        }
    ],
    "highestFieldId": 2,
    "partitionKeys": [],
    "primaryKeys": [
        "f0"
    ],
    "options": {
        "bucket": "1",
        "file.format": "orc",
        "orc.filter.use.selected": "true",
        "orc.sarg.to.filter": "true",
        "target-file-size": "1024 MB",
        "file.block-size": "8 MB",
        "deletion-vectors.enabled": "true"
    },
    "timeMillis": 1731654078602
}

java code

import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.junit.Test;

import java.util.List;
import java.util.Random;
import java.util.UUID;

public class SpeedOrcLookupNoPrimaryKey {
    @Test
    public void testWriter() throws Exception {
        writer();
    }
    @Test
    public void testReader() throws Exception {
        reader();
    }
    private void reader() throws Exception {
        Table table =  // getTable();
        PredicateBuilder builder = new PredicateBuilder(
                RowType.of(DataTypes.INT(),DataTypes.INT(),
                        DataTypes.STRING(),
                        DataTypes.STRING()));

        int[] projection = new int[] {0, 1, 2,3};

        ReadBuilder readBuilder = table.newReadBuilder()
                .withProjection(projection);

        Random random = new Random();
        for(int i = 0 ; i < 30 ; i ++){
            InnerTableRead read = (InnerTableRead)readBuilder.newRead();
            int key = random.nextInt(4000000);

            Predicate keyFilter = builder.equal(1, 2903308);

            InnerTableScan tableScan = (InnerTableScan) readBuilder
                    .withFilter(keyFilter)
                    .newScan();
            InnerTableScan innerTableScan = tableScan.withFilter(keyFilter);
            TableScan.Plan plan = innerTableScan.plan();
            List<Split> splits = plan.splits();

            read.withFilter(keyFilter);//.executeFilter();
            RecordReader<InternalRow> reader = read.createReader(splits);

            reader.forEachRemaining(internalRow -> {

                int f0 = internalRow.getInt(0);
                int f1 = internalRow.getInt(1);
                String f2 = internalRow.getString(2).toString();
                String f3 = internalRow.getString(3).toString();
            });
        }
        long startTime = System.currentTimeMillis();

        for(int i = 0 ; i < 1000 ; i ++){
            InnerTableRead read = (InnerTableRead)readBuilder.newRead();
            int key = random.nextInt(4000000);

            Predicate keyFilter = builder.equal(1, key);

            InnerTableScan tableScan = (InnerTableScan) readBuilder
                    .withFilter(keyFilter)
                    .newScan();
            InnerTableScan innerTableScan = tableScan.withFilter(keyFilter);
            TableScan.Plan plan = innerTableScan.plan();
            List<Split> splits = plan.splits();

            read.withFilter(keyFilter);//.executeFilter();
            RecordReader<InternalRow> reader = read.createReader(splits);

            reader.forEachRemaining(internalRow -> {

                int f0 = internalRow.getInt(0);
                int f1 = internalRow.getInt(1);
                String f2 = internalRow.getString(2).toString();
                String f3 = internalRow.getString(3).toString();
            });
        }
        long stopTime = System.currentTimeMillis();
        System.out.println("time : " + (stopTime - startTime));
    }
    private void writer() throws Exception {
        Table table =  // getTable();

        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder()
                ;
        BatchTableWrite write = writeBuilder
                .newWrite() .withIOManager(new IOManagerImpl("./tmp")).asBatchTableWrite();  // TableWriteImpl

        Random random = new Random();

        long startTime = System.currentTimeMillis();

        for(int i = 0; i < 1000_0000; i++){

            GenericRow genericRow = GenericRow.of(
                    i,
                    random.nextInt(1000_0000),
                    BinaryString.fromString(String.valueOf(random.nextInt())),
                    BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
                    ,BinaryString.fromString(UUID.randomUUID().toString())
            );
            write.write(genericRow);
        }

        List<CommitMessage> messages = write.prepareCommit();

        BatchTableCommit commit = writeBuilder.newCommit();

        commit.commit(messages);

        long stopTime = System.currentTimeMillis();

        System.out.println("time : " + (stopTime - startTime));
    }
}

@Tan-JiaLiang

[JingsongLi]

@Tan-JiaLiang Very thanks for your detailed benchmark!

[Tan-JiaLiang]

Hi, @ranxianglei.
The min/max index combined with the page-grained bloom-filter is very useful in the case of EQ predicates. But it is only useful in the EQ predicate case.
More importantly, I wanted to explore the ability to use some indices to optimise other predicates, such as the Range and the TopK search.

Using EQ as the benchmark predicate is just because of I think it is simple and it can more clearly compare the effect and side effect after push down.

[ranxianglei]

min max can also be used for ranges, not just eq. In addition, queries outside the sorting optimization range may not achieve good results. It is recommended to first check whether there is still a problem with Parquet's pushdown implementation. The benefits of fixing a Parquet pushdown problem may be much greater than adding an index. @Tan-JiaLiang

[Tan-JiaLiang]

Hi, @ranxianglei

The benefits of fixing a Parquet pushdown problem may be much greater than adding an index.

I quite agree with this view. But I do not see of any room for improvement temporarily, since it's already optimised for page-level filtering with the zonemap index. But if there is, I'd love to look into it.

min max can also be used for ranges, not just eq.

Yes, min/max can do filtering in range case, but the bloom-filter can not. The min/max index is always has the false-positive and it is hard to get good results without sort compact. And the sort compact can not sort on too many fields.

More: the difference between the bitmap index and the zone map index (min/max index)

The bitmap category index is the position index and it does not have the false-positive problem and can do more in other predicate case. However, it is larger than the zonemap index, and loading and evaluation is slower than the zonemap index.

The main of this PR is to combine the selection and the deletion result and push it down and do the page filtering for better performance.

[Tan-JiaLiang]

Because of only push down the deletion vector result does not work in the benchmark, this PR will change to merge the file index and deletion vector and push down. cc @JingsongLi

[Tan-JiaLiang]

@JingsongLi Can you please take a look again? If something wrong please feel free to say.

[JingsongLi]

+1