You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/29 10:13:13 UTC
[GitHub] [flink-table-store] LadyForest opened a new pull request, #280: [FLINK-29129] Try best to push down value filters
LadyForest opened a new pull request, #280:
URL: https://github.com/apache/flink-table-store/pull/280
As an improvement, value filters can be pushed down when generated splits do not have key range overlap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] LadyForest closed pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
LadyForest closed pull request #280: [FLINK-29129] Try best to push down value filters
URL: https://github.com/apache/flink-table-store/pull/280
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r959231586
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java:
##########
@@ -87,34 +108,78 @@ public KeyValueFileStoreRead withValueProjection(int[][] projectedFields) {
@Override
public FileStoreRead<KeyValue> withFilter(Predicate predicate) {
- this.filters = splitAnd(predicate);
+ allFilters = new ArrayList<>();
+ List<String> primaryKeys = tableSchema.trimmedPrimaryKeys();
+ Set<String> nonPrimaryKeys =
+ tableSchema.fieldNames().stream()
+ .filter(name -> !primaryKeys.contains(name))
+ .collect(Collectors.toSet());
+ for (Predicate sub : splitAnd(predicate)) {
+ allFilters.add(sub);
+ if (!containsFields(sub, nonPrimaryKeys)) {
+ if (keyFilters == null) {
+ keyFilters = new ArrayList<>();
+ }
+ // TODO Actually, the index is wrong, but it is OK.
+ // The orc filter just use name instead of index.
+ keyFilters.add(sub);
+ }
+ }
return this;
}
@Override
public RecordReader<KeyValue> createReader(Split split) throws IOException {
if (split.isIncremental()) {
- DataFileReader dataFileReader =
- dataFileReaderFactory.create(split.partition(), split.bucket(), true, filters);
+ // incremental mode cannot push down value filters, because the update for the same key
+ // may occur in the next split
+ DataFileReader dataFileReader = createDataFileReader(split, true, false);
// Return the raw file contents without merging
List<ConcatRecordReader.ReaderSupplier<KeyValue>> suppliers = new ArrayList<>();
for (DataFileMeta file : split.files()) {
- suppliers.add(
- () -> dataFileReader.read(changelogFile(file).orElse(file.fileName())));
+ if (acceptFilter(false).test(file)) {
+ suppliers.add(
+ () -> dataFileReader.read(changelogFile(file).orElse(file.fileName())));
+ }
}
return ConcatRecordReader.create(suppliers);
} else {
// in this case merge tree should merge records with same key
// Do not project key in MergeTreeReader.
- DataFileReader dataFileReader =
- dataFileReaderFactory.create(split.partition(), split.bucket(), false, filters);
- MergeTreeReader reader =
- new MergeTreeReader(
- new IntervalPartition(split.files(), keyComparator).partition(),
- true,
- dataFileReader,
- keyComparator,
- mergeFunction.copy());
+ List<List<SortedRun>> sections =
+ new IntervalPartition(split.files(), keyComparator).partition();
+ DataFileReader dataFileReaderWithAllFilters = createDataFileReader(split, false, true);
+ DataFileReader dataFileReaderWithKeyFilters = createDataFileReader(split, false, false);
+ MergeFunction mergeFunc = mergeFunction.copy();
+ List<ConcatRecordReader.ReaderSupplier<KeyValue>> readers = new ArrayList<>();
+ for (List<SortedRun> section : sections) {
+ // if key ranges do not have overlap, value filter can be pushed down as well
+ boolean acceptAll = section.size() == 1;
+ List<SortedRun> hitSection = new ArrayList<>();
+ for (SortedRun run : section) {
+ List<DataFileMeta> hitFiles = new ArrayList<>();
+ for (DataFileMeta file : run.files()) {
Review Comment:
Maybe use `Stream.filter` is better to read.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r959229043
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java:
##########
@@ -50,13 +60,18 @@
*/
public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
+ private final TableSchema tableSchema;
private final DataFileReader.Factory dataFileReaderFactory;
private final Comparator<RowData> keyComparator;
private final MergeFunction mergeFunction;
+ private final boolean valueCountMode;
+ private final FieldStatsArraySerializer rowStatsConverter;
+
+ @Nullable private int[][] keyProjectedFields;
- private int[][] keyProjectedFields;
+ @Nullable private List<Predicate> keyFilters;
- private List<Predicate> filters;
+ @Nullable private List<Predicate> allFilters;
Review Comment:
Introduce two filter:
```
private List<Predicate> overlappedFilters;
private List<Predicate> nonOverlappedFilters;
```
In the `acceptAllFilter`, distinguish whether they are `overlappedFilters` or `overlappedFilters` according to the `valueCountMode` and key or all.
So in the `createReader`, there is no need to judge `acceptAllFilter`.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java:
##########
@@ -50,13 +60,18 @@
*/
public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
+ private final TableSchema tableSchema;
private final DataFileReader.Factory dataFileReaderFactory;
private final Comparator<RowData> keyComparator;
private final MergeFunction mergeFunction;
+ private final boolean valueCountMode;
+ private final FieldStatsArraySerializer rowStatsConverter;
+
+ @Nullable private int[][] keyProjectedFields;
- private int[][] keyProjectedFields;
+ @Nullable private List<Predicate> keyFilters;
- private List<Predicate> filters;
+ @Nullable private List<Predicate> allFilters;
Review Comment:
Introduce two filter:
```
private List<Predicate> overlappedFilters;
private List<Predicate> nonOverlappedFilters;
```
In the `acceptAllFilter`, distinguish whether they are `overlappedFilters` or `overlappedFilters` according to the `valueCountMode` and key or all.
So in the `createReader`, there is no need to judge `acceptAllFilter`.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r958013297
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java:
##########
@@ -87,15 +94,28 @@ public KeyValueFileStoreRead withValueProjection(int[][] projectedFields) {
@Override
public FileStoreRead<KeyValue> withFilter(Predicate predicate) {
- this.filters = splitAnd(predicate);
+ this.keyFilters = splitAnd(predicate);
+ return this;
+ }
+
+ public FileStoreRead<KeyValue> withValueFilter(Predicate predicate) {
+ this.valueFilters = splitAnd(predicate);
return this;
}
@Override
public RecordReader<KeyValue> createReader(Split split) throws IOException {
+ List<List<SortedRun>> sections =
+ new IntervalPartition(split.files(), keyComparator).partition();
if (split.isIncremental()) {
- DataFileReader dataFileReader =
- dataFileReaderFactory.create(split.partition(), split.bucket(), true, filters);
+ boolean keyRangeOverlap = false;
+ for (List<SortedRun> section : sections) {
Review Comment:
Maybe we can just filter all for incremental?
Because incremental does not have a merge when it is read, it seems that it is OK to directly filter and push down?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi merged pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi merged PR #280:
URL: https://github.com/apache/flink-table-store/pull/280
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r959229043
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java:
##########
@@ -50,13 +60,18 @@
*/
public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
+ private final TableSchema tableSchema;
private final DataFileReader.Factory dataFileReaderFactory;
private final Comparator<RowData> keyComparator;
private final MergeFunction mergeFunction;
+ private final boolean valueCountMode;
+ private final FieldStatsArraySerializer rowStatsConverter;
+
+ @Nullable private int[][] keyProjectedFields;
- private int[][] keyProjectedFields;
+ @Nullable private List<Predicate> keyFilters;
- private List<Predicate> filters;
+ @Nullable private List<Predicate> allFilters;
Review Comment:
Introduce two filter:
```
private List<Predicate> overlappedFilters;
private List<Predicate> nonOverlappedFilters;
```
So in the `createReader`, there is no need to judge `acceptAllFilter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r958013297
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java:
##########
@@ -87,15 +94,28 @@ public KeyValueFileStoreRead withValueProjection(int[][] projectedFields) {
@Override
public FileStoreRead<KeyValue> withFilter(Predicate predicate) {
- this.filters = splitAnd(predicate);
+ this.keyFilters = splitAnd(predicate);
+ return this;
+ }
+
+ public FileStoreRead<KeyValue> withValueFilter(Predicate predicate) {
+ this.valueFilters = splitAnd(predicate);
return this;
}
@Override
public RecordReader<KeyValue> createReader(Split split) throws IOException {
+ List<List<SortedRun>> sections =
+ new IntervalPartition(split.files(), keyComparator).partition();
if (split.isIncremental()) {
- DataFileReader dataFileReader =
- dataFileReaderFactory.create(split.partition(), split.bucket(), true, filters);
+ boolean keyRangeOverlap = false;
+ for (List<SortedRun> section : sections) {
Review Comment:
Maybe we can just filter all for incremental?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] LadyForest closed pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
LadyForest closed pull request #280: [FLINK-29129] Try best to push down value filters
URL: https://github.com/apache/flink-table-store/pull/280
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r958013575
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java:
##########
@@ -164,19 +164,22 @@ public TableRead newRead() {
@Override
public TableRead withFilter(Predicate predicate) {
- List<Predicate> predicates = new ArrayList<>();
+ List<Predicate> keyPredicates = new ArrayList<>();
+ List<Predicate> valuePredicates = new ArrayList<>();
for (Predicate sub : splitAnd(predicate)) {
- // TODO support value filter
if (containsFields(sub, nonPrimaryKeys)) {
- continue;
+ valuePredicates.add(sub);
Review Comment:
key only -> key predicates
key + value -> all predicates
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r958015488
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java:
##########
@@ -106,15 +126,24 @@ public RecordReader<KeyValue> createReader(Split split) throws IOException {
} else {
// in this case merge tree should merge records with same key
// Do not project key in MergeTreeReader.
- DataFileReader dataFileReader =
- dataFileReaderFactory.create(split.partition(), split.bucket(), false, filters);
- MergeTreeReader reader =
- new MergeTreeReader(
- new IntervalPartition(split.files(), keyComparator).partition(),
- true,
- dataFileReader,
- keyComparator,
- mergeFunction.copy());
+ DataFileReader dataFileReaderWithAllFilters = createDataFileReader(split, false, true);
+ DataFileReader dataFileReaderWithKeyFilters = createDataFileReader(split, false, false);
+ MergeFunction mergeFunc = mergeFunction.copy();
+ List<ConcatRecordReader.ReaderSupplier<KeyValue>> readers = new ArrayList<>();
+ for (List<SortedRun> section : sections) {
+ DataFileReader dataFileReader;
+ if (section.size() == 1) {
+ // key ranges do not overlap, and value filters can be pushed down
+ dataFileReader = dataFileReaderWithAllFilters;
+ } else {
+ dataFileReader = dataFileReaderWithKeyFilters;
+ }
+ readers.add(
Review Comment:
Maybe we can do more, we can filter the files by `Predicate.test`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r958013575
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java:
##########
@@ -164,19 +164,22 @@ public TableRead newRead() {
@Override
public TableRead withFilter(Predicate predicate) {
- List<Predicate> predicates = new ArrayList<>();
+ List<Predicate> keyPredicates = new ArrayList<>();
+ List<Predicate> valuePredicates = new ArrayList<>();
for (Predicate sub : splitAnd(predicate)) {
- // TODO support value filter
if (containsFields(sub, nonPrimaryKeys)) {
- continue;
+ valuePredicates.add(sub);
Review Comment:
There are two cases only:
1. key only -> key predicates
2. key + value -> all predicates
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r959226860
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java:
##########
@@ -62,6 +62,12 @@ public MergeTreeReader(
this.reader = ConcatRecordReader.create(readers);
}
+ public MergeTreeReader(boolean dropDelete, List<ReaderSupplier<KeyValue>> readers)
Review Comment:
`sectionReaders`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #280:
URL: https://github.com/apache/flink-table-store/pull/280#discussion_r958012117
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java:
##########
@@ -164,19 +164,22 @@ public TableRead newRead() {
@Override
public TableRead withFilter(Predicate predicate) {
- List<Predicate> predicates = new ArrayList<>();
+ List<Predicate> keyPredicates = new ArrayList<>();
+ List<Predicate> valuePredicates = new ArrayList<>();
for (Predicate sub : splitAnd(predicate)) {
- // TODO support value filter
if (containsFields(sub, nonPrimaryKeys)) {
- continue;
+ valuePredicates.add(sub);
Review Comment:
We should keep all predicates when we can do predicate push down for values.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] LadyForest closed pull request #280: [FLINK-29129] Try best to push down value filters
Posted by GitBox <gi...@apache.org>.
LadyForest closed pull request #280: [FLINK-29129] Try best to push down value filters
URL: https://github.com/apache/flink-table-store/pull/280
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org