[Data] refactor checkpointfilter with anti-join

[my-vegetable-has-exploded]

Description

Currently, each ReadTask initializes a checkpointer to filter rows whose primary key already exists in checkpoint_id (similar to a broadcast join). This is a typical anti-join pattern. I think that we refactor the checkpoint filter using an anti-join approach.

By adopting anti-join, the checkpoint filter may benefit from optimizer rules:
1.The optimizer can choose between broadcast join or distributed hash join based on dataset statistics, potentially resolving efficiency or memory issues (such as #60294).
2.Once the runtimefilter/dynamic optimizer is supported, the checkpoint filter may be pushed down to the source side.

However, currently the hash join is based on ShuffleAggregation, which remains a pipeline broker. This architecture blocks the streaming execution pipeline, as all data must be completely shuffled and materialized before downstream operators can begin processing, thereby preventing true pipelined execution.

ptal @wxwmd @owenowenisme

[wxwmd]

i think current filter has been pushed down to the source side(ReadTask), post some architecture ptotos will help me to understand

[my-vegetable-has-exploded]

i think current filter has been pushed down to the source side(ReadTask), post some architecture ptotos will help me to understand

hi @wxwmd, checkpoint filtering happens after the data has already been read. I was originally hoping to use not in checkpoint_id filter on the IO side—like skipping certain row groups or pages when reading Parquet or Iceberg. But after doing some research, I found that most scanners don't handle not in list filter very well when size of the list is large. So the second point may make no sense.