[core] Push bucket filter into manifest entry reader (#4258)

paimon-common/src/main/java/org/apache/paimon/utils/BiFilter.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.utils;
+
+import org.apache.paimon.predicate.Predicate;
+
+/**
+ * Represents a filter (boolean-valued function) of two argument. This class is for avoiding name
+ * conflicting to {@link Predicate}.
+ */
+@FunctionalInterface
+public interface BiFilter<T, U> {
+
+ BiFilter<?, ?> ALWAYS_TRUE = (t, u) -> true;
+
+ boolean test(T t, U u);
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ static <T, U> BiFilter<T, U> alwaysTrue() {
+ return (BiFilter) ALWAYS_TRUE;
+ }
+}

paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java

@@ -27,8 +27,8 @@
 import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
 import org.apache.paimon.operation.AppendOnlyFixedBucketFileStoreWrite;
 import org.apache.paimon.operation.AppendOnlyUnawareBucketFileStoreWrite;
+import org.apache.paimon.operation.BucketSelectConverter;
 import org.apache.paimon.operation.RawFileSplitRead;
-import org.apache.paimon.operation.ScanBucketFilter;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
@@ -38,6 +38,7 @@
 
 import java.util.Comparator;
 import java.util.List;
+import java.util.Optional;
 
 import static org.apache.paimon.predicate.PredicateBuilder.and;
 import static org.apache.paimon.predicate.PredicateBuilder.pickTransformFieldMapping;
@@ -106,6 +107,7 @@ public AppendOnlyFileStoreWrite newWrite(
  newRead(),
  schema.id(),
  rowType,
+ partitionType,
  pathFactory(),
  snapshotManager(),
  newScan(true).withManifestCacheFilter(manifestFilter),
@@ -119,6 +121,7 @@ public AppendOnlyFileStoreWrite newWrite(
  schema.id(),
  commitUser,
  rowType,
+ partitionType,
  pathFactory(),
  snapshotManager(),
  newScan(true).withManifestCacheFilter(manifestFilter),
@@ -129,39 +132,34 @@ public AppendOnlyFileStoreWrite newWrite(
  }
 
  private AppendOnlyFileStoreScan newScan(boolean forWrite) {
- ScanBucketFilter bucketFilter =
- new ScanBucketFilter(bucketKeyType) {
- @Override
- public void pushdown(Predicate predicate) {
- if (bucketMode() != BucketMode.HASH_FIXED) {
- return;
- }
-
- if (bucketKeyType.getFieldCount() == 0) {
- return;
- }
-
- List<Predicate> bucketFilters =
- pickTransformFieldMapping(
- splitAnd(predicate),
- rowType.getFieldNames(),
- bucketKeyType.getFieldNames());
- if (!bucketFilters.isEmpty()) {
- setBucketKeyFilter(and(bucketFilters));
- }
+ BucketSelectConverter bucketSelectConverter =
+ predicate -> {
+ if (bucketMode() != BucketMode.HASH_FIXED) {
+ return Optional.empty();
  }
+
+ if (bucketKeyType.getFieldCount() == 0) {
+ return Optional.empty();
+ }
+
+ List<Predicate> bucketFilters =
+ pickTransformFieldMapping(
+ splitAnd(predicate),
+ rowType.getFieldNames(),
+ bucketKeyType.getFieldNames());
+ if (!bucketFilters.isEmpty()) {
+ return BucketSelectConverter.create(and(bucketFilters), bucketKeyType);
+ }
+ return Optional.empty();
  };
 
  return new AppendOnlyFileStoreScan(
  newManifestsReader(forWrite),
- partitionType,
- bucketFilter,
+ bucketSelectConverter,
  snapshotManager(),
  schemaManager,
  schema,
  manifestFileFactory(forWrite),
- options.bucket(),
- forWrite,
  options.scanManifestParallelism(),
  options.fileIndexReadEnabled());
  }

paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java

@@ -28,11 +28,11 @@
 import org.apache.paimon.io.KeyValueFileReaderFactory;
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
+import org.apache.paimon.operation.BucketSelectConverter;
 import org.apache.paimon.operation.KeyValueFileStoreScan;
 import org.apache.paimon.operation.KeyValueFileStoreWrite;
 import org.apache.paimon.operation.MergeFileSplitRead;
 import org.apache.paimon.operation.RawFileSplitRead;
-import org.apache.paimon.operation.ScanBucketFilter;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.schema.KeyValueFieldsExtractor;
 import org.apache.paimon.schema.SchemaManager;
@@ -50,6 +50,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.function.Supplier;
 
@@ -210,35 +211,30 @@ private Map<String, FileStorePathFactory> format2PathFactory() {
  }
 
  private KeyValueFileStoreScan newScan(boolean forWrite) {
- ScanBucketFilter bucketFilter =
- new ScanBucketFilter(bucketKeyType) {
- @Override
- public void pushdown(Predicate keyFilter) {
- if (bucketMode() != BucketMode.HASH_FIXED) {
- return;
- }
-
- List<Predicate> bucketFilters =
- pickTransformFieldMapping(
- splitAnd(keyFilter),
- keyType.getFieldNames(),
- bucketKeyType.getFieldNames());
- if (bucketFilters.size() > 0) {
- setBucketKeyFilter(and(bucketFilters));
- }
+ BucketSelectConverter bucketSelectConverter =
+ keyFilter -> {
+ if (bucketMode() != BucketMode.HASH_FIXED) {
+ return Optional.empty();
  }
+
+ List<Predicate> bucketFilters =
+ pickTransformFieldMapping(
+ splitAnd(keyFilter),
+ keyType.getFieldNames(),
+ bucketKeyType.getFieldNames());
+ if (bucketFilters.size() > 0) {
+ return BucketSelectConverter.create(and(bucketFilters), bucketKeyType);
+ }
+ return Optional.empty();
  };
  return new KeyValueFileStoreScan(
  newManifestsReader(forWrite),
- partitionType,
- bucketFilter,
+ bucketSelectConverter,
  snapshotManager(),
  schemaManager,
  schema,
  keyValueFieldsExtractor,
  manifestFileFactory(forWrite),
- options.bucket(),
- forWrite,
  options.scanManifestParallelism(),
  options.deletionVectorsEnabled(),
  options.mergeEngine(),

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySerializer.java

@@ -96,4 +96,8 @@ public static Function<InternalRow, Integer> totalBucketGetter() {
  public static Function<InternalRow, String> fileNameGetter() {
  return row -> row.getRow(5, DataFileMeta.SCHEMA.getFieldCount()).getString(0).toString();
  }
+
+ public static Function<InternalRow, Integer> levelGetter() {
+ return row -> row.getRow(5, DataFileMeta.SCHEMA.getFieldCount()).getInt(10);
+ }
 }