This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6d06143b4e [core] Execute complete filters with unprojected fields
(#9872)
6d06143b4e is described below
commit 6d06143b4e4deb4df8fcb373974f7ad198882ede
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Sep 16 13:07:25 2026 +0800
[core] Execute complete filters with unprojected fields (#9872)
---
.../apache/paimon/predicate/PredicateRemapper.java | 87 +++++
.../java/org/apache/paimon/utils/TypeUtils.java | 19 ++
.../paimon/predicate/PredicateRemapperTest.java | 144 ++++++++
.../paimon/catalog/TableQueryAuthResult.java | 133 +++-----
.../operation/DataEvolutionFileStoreScan.java | 17 +-
.../paimon/table/format/FormatReadBuilder.java | 18 +-
.../paimon/table/format/FormatTableRead.java | 40 +--
.../paimon/table/source/AbstractDataTableRead.java | 199 ++---------
.../paimon/table/source/AbstractDataTableScan.java | 56 ++--
.../apache/paimon/table/source/ReadTransform.java | 178 ++++++++++
.../org/apache/paimon/table/source/TableRead.java | 12 +
.../org/apache/paimon/rest/RESTCatalogTest.java | 87 +++++
.../paimon/table/AppendOnlySimpleTableTest.java | 10 +-
.../paimon/table/DataEvolutionFileIndexTest.java | 44 ++-
.../paimon/table/PrimaryKeySimpleTableTest.java | 26 ++
.../paimon/table/format/FormatTableReadTest.java | 109 ++++++
.../table/source/AbstractDataTableReadTest.java | 369 ++++++++++++++++++++-
.../spark/procedure/PermissionProcedureTest.scala | 4 +-
18 files changed, 1197 insertions(+), 355 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateRemapper.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateRemapper.java
new file mode 100644
index 0000000000..8dc9d8b0f1
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateRemapper.java
@@ -0,0 +1,87 @@
+/*
+ * 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.predicate;
+
+import org.apache.paimon.types.RowType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Resolves field references by name to positional indices and types in a read
schema. Every
+ * referenced field must exist; unresolved fields and malformed compound
predicates are rejected.
+ */
+public final class PredicateRemapper implements PredicateVisitor<Predicate> {
+
+ private final RowType rowType;
+
+ private PredicateRemapper(RowType rowType) {
+ this.rowType = rowType;
+ }
+
+ public static Predicate remap(Predicate predicate, RowType rowType) {
+ return predicate.visit(new PredicateRemapper(rowType));
+ }
+
+ public static Transform remap(Transform transform, RowType rowType) {
+ return transform.copyWithNewInputs(new
PredicateRemapper(rowType).remapInputs(transform));
+ }
+
+ private List<Object> remapInputs(Transform transform) {
+ List<Object> inputs = new ArrayList<>();
+ for (Object input : transform.inputs()) {
+ if (input instanceof FieldRef) {
+ FieldRef field = (FieldRef) input;
+ int index = rowType.getFieldIndex(field.name());
+ checkArgument(
+ index >= 0,
+ "Cannot resolve field '%s' in read schema %s.",
+ field.name(),
+ rowType);
+ inputs.add(new FieldRef(index, field.name(),
rowType.getTypeAt(index)));
+ } else {
+ inputs.add(input);
+ }
+ }
+ return inputs;
+ }
+
+ @Override
+ public Predicate visit(LeafPredicate predicate) {
+ return predicate.copyWithNewInputs(remapInputs(predicate.transform()));
+ }
+
+ @Override
+ public Predicate visit(CompoundPredicate predicate) {
+ checkArgument(predicate.function() != null, "Compound predicate
function cannot be null.");
+ checkArgument(predicate.children() != null, "Compound predicate
children cannot be null.");
+ checkArgument(
+ !predicate.children().isEmpty(), "Compound predicate must
contain a predicate.");
+ List<Predicate> children = new ArrayList<>();
+ for (Predicate child : predicate.children()) {
+ checkArgument(child != null, "Compound predicate child cannot be
null.");
+ children.add(child.visit(this));
+ }
+ return children.size() == 1
+ ? children.get(0)
+ : new CompoundPredicate(predicate.function(), children);
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
b/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
index cfe0f95afb..6fdc97ecde 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
@@ -54,6 +54,7 @@ import java.util.Base64;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.TimeZone;
import java.util.stream.Collectors;
@@ -91,6 +92,24 @@ public class TypeUtils {
.collect(Collectors.toList()));
}
+ /**
+ * Append required fields available in the table schema to a read type.
Existing fields,
+ * including nested projections, are preserved. Returns the original read
type if unchanged.
+ */
+ public static RowType withMissingFields(
+ RowType tableType, RowType readType, Set<String> requiredFields) {
+ List<DataField> fields = null;
+ for (DataField field : tableType.getFields()) {
+ if (requiredFields.contains(field.name()) &&
!readType.containsField(field.name())) {
+ if (fields == null) {
+ fields = new ArrayList<>(readType.getFields());
+ }
+ fields.add(field);
+ }
+ }
+ return fields == null ? readType : readType.copy(fields);
+ }
+
public static Object castFromString(String s, DataType type) {
return castFromStringInternal(s, type, false);
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateRemapperTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateRemapperTest.java
new file mode 100644
index 0000000000..e9f338427c
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateRemapperTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.predicate;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests exact predicate binding by name, including transforms and missing
operands. */
+class PredicateRemapperTest {
+
+ private static final RowType TYPE =
+ RowType.of(
+ new DataField(17, "a", DataTypes.STRING()),
+ new DataField(41, "b", DataTypes.STRING()),
+ new DataField(71, "c", DataTypes.INT()));
+
+ @Test
+ void testCompoundPredicateOnReorderedFields() {
+ PredicateBuilder builder = new PredicateBuilder(TYPE);
+ Predicate predicate =
+ PredicateBuilder.and(
+ builder.equal(0, BinaryString.fromString("left")),
+ PredicateBuilder.or(
+ builder.equal(1,
BinaryString.fromString("right")),
+ builder.greaterThan(2, 1)));
+ Predicate remapped = PredicateRemapper.remap(predicate,
TYPE.project("b", "c", "a"));
+ assertThat(
+ remapped.test(
+ GenericRow.of(
+ BinaryString.fromString("right"),
+ 0,
+ BinaryString.fromString("left"))))
+ .isTrue();
+ assertThat(
+ remapped.test(
+ GenericRow.of(
+ BinaryString.fromString("other"),
+ 2,
+ BinaryString.fromString("left"))))
+ .isTrue();
+ assertThat(
+ remapped.test(
+ GenericRow.of(
+ BinaryString.fromString("other"),
+ 0,
+ BinaryString.fromString("left"))))
+ .isFalse();
+ assertThat(
+ remapped.test(
+ GenericRow.of(
+ BinaryString.fromString("right"),
+ 2,
+ BinaryString.fromString("wrong"))))
+ .isFalse();
+ // Binding must not mutate the original predicate or its positional
references.
+ assertThat(
+ predicate.test(
+ GenericRow.of(
+ BinaryString.fromString("left"),
+ BinaryString.fromString("right"),
+ 0)))
+ .isTrue();
+ }
+
+ @Test
+ void testFieldIdsAndLiteralTransformInputs() {
+ Transform transform =
+ new ConcatWsTransform(
+ Arrays.asList(
+ BinaryString.fromString("-"),
+ new FieldRef(17, "a", DataTypes.STRING()),
+ new FieldRef(41, "b", DataTypes.STRING())));
+ RowType readType = TYPE.project("b", "a");
+ GenericRow row =
+ GenericRow.of(BinaryString.fromString("right"),
BinaryString.fromString("left"));
+ assertThat(PredicateRemapper.remap(transform, readType).transform(row))
+ .isEqualTo(BinaryString.fromString("left-right"));
+ Predicate predicate =
+ new PredicateBuilder(TYPE).equal(transform,
BinaryString.fromString("left-right"));
+ assertThat(PredicateRemapper.remap(predicate,
readType).test(row)).isTrue();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMissingConjunctOrDisjunctIsRejected(boolean and) {
+ PredicateBuilder builder = new PredicateBuilder(TYPE);
+ Predicate a = builder.isNotNull(0);
+ Predicate b = builder.isNotNull(1);
+ Predicate predicate = and ? PredicateBuilder.and(a, b) :
PredicateBuilder.or(a, b);
+ assertThatThrownBy(() -> PredicateRemapper.remap(predicate,
TYPE.project("a")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot resolve field 'b'");
+ }
+
+ @Test
+ void testNullPredicateOnReorderedFields() {
+ Predicate predicate = new PredicateBuilder(TYPE).isNull(0);
+ Predicate remapped = PredicateRemapper.remap(predicate,
TYPE.project("b", "a"));
+ assertThat(remapped.test(GenericRow.of(BinaryString.fromString("b"),
null))).isTrue();
+ assertThat(remapped.test(GenericRow.of(null,
BinaryString.fromString("a")))).isFalse();
+ }
+
+ @Test
+ void testMalformedCompoundIsRejected() {
+ for (String json :
+ Arrays.asList(
+
"{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[]}",
+
"{\"kind\":\"COMPOUND\",\"function\":null,\"children\":[]}",
+
"{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[null]}")) {
+ Predicate predicate = JsonSerdeUtil.fromJson(json,
Predicate.class);
+ assertThatThrownBy(() -> PredicateRemapper.remap(predicate, TYPE))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java
index e3fd81442b..6ab47711dd 100644
---
a/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java
+++
b/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java
@@ -22,10 +22,9 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.predicate.And;
import org.apache.paimon.predicate.CompoundPredicate;
-import org.apache.paimon.predicate.FieldRef;
-import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.PredicateRemapper;
import org.apache.paimon.predicate.PredicateVisitor;
import org.apache.paimon.predicate.Transform;
import org.apache.paimon.reader.RecordReader;
@@ -43,6 +42,7 @@ import org.apache.paimon.utils.InternalRowUtils;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.ListUtils;
import org.apache.paimon.utils.StringUtils;
+import org.apache.paimon.utils.TypeUtils;
import javax.annotation.Nullable;
@@ -186,16 +186,8 @@ public class TableQueryAuthResult implements Serializable {
@Nullable
public static RowType appendMissingFields(
RowType tableType, RowType readType, Set<String> ruleFields) {
- List<DataField> widenedFields = null;
- for (DataField field : tableType.getFields()) {
- if (ruleFields.contains(field.name()) &&
!readType.containsField(field.name())) {
- if (widenedFields == null) {
- widenedFields = new ArrayList<>(readType.getFields());
- }
- widenedFields.add(field);
- }
- }
- return widenedFields == null ? null : readType.copy(widenedFields);
+ RowType widened = TypeUtils.withMissingFields(tableType, readType,
ruleFields);
+ return widened == readType ? null : widened;
}
public TableScan.Plan convertPlan(TableScan.Plan plan) {
@@ -246,7 +238,7 @@ public class TableQueryAuthResult implements Serializable {
*/
@Nullable
public static Predicate remapPredicate(Predicate predicate, RowType
rowType) {
- return predicate.visit(new PredicateRemapper(rowType));
+ return PredicateRemapper.remap(predicate, rowType);
}
public Map<String, Transform> extractColumnMasking() {
@@ -415,6 +407,45 @@ public class TableQueryAuthResult implements Serializable {
tableType.getFieldNames());
}
+ /** Validate a physical projection before adding columns required by
authorization rules. */
+ public void validateReadType(
+ RowType tableType,
+ RowType readType,
+ Set<String> ruleFields,
+ Set<String> resolvedBlobViewFields) {
+ Set<String> maskTargets = extractColumnMasking().keySet();
+ for (String name : readType.getFieldNames()) {
+ if (!ruleFields.contains(name) && !maskTargets.contains(name)) {
+ continue;
+ }
+ if (!tableType.containsField(name)) {
+ continue;
+ }
+ // rules must not touch a nested-pruned column (partial value)
+ DataField tableField = tableType.getField(name);
+ if (!readType.getField(name).type().equals(tableField.type())) {
+ throw new IllegalStateException(
+ String.format(
+ "Query auth rules involve column '%s', which
the query "
+ + "projects with a pruned type %s
instead of its "
+ + "table type %s; cannot apply the
rules to a "
+ + "partial column.",
+ name, readType.getField(name).type(),
tableField.type()));
+ }
+ }
+ for (String name : ruleFields) {
+ if (resolvedBlobViewFields.contains(name) &&
!readType.containsField(name)) {
+ // auth-added columns bypass blob-view resolution
+ throw new IllegalStateException(
+ String.format(
+ "Query auth rules read blob-view column '%s',
which the query "
+ + "does not project; such columns
cannot be resolved. "
+ + "Project the column or adjust the
rule.",
+ name));
+ }
+ }
+ }
+
/**
* Applies the row filter and column masking to {@code reader}. Rules are
remapped by name;
* masks apply only to targets in {@code activeFields}, the columns
readable from the query.
@@ -481,82 +512,8 @@ public class TableQueryAuthResult implements Serializable {
targetColumn,
outputRowType);
- List<Object> newInputs = new ArrayList<>();
- for (Object input : transform.inputs()) {
- if (input instanceof FieldRef) {
- FieldRef ref = (FieldRef) input;
- int newIndex = outputRowType.getFieldIndex(ref.name());
- if (newIndex < 0) {
- throw new IllegalArgumentException(
- "Column masking refers to field '"
- + ref.name()
- + "' which is not present in output
row type "
- + outputRowType);
- }
- DataType type = outputRowType.getTypeAt(newIndex);
- newInputs.add(new FieldRef(newIndex, ref.name(), type));
- } else {
- newInputs.add(input);
- }
- }
- out.put(targetIndex, transform.copyWithNewInputs(newInputs));
+ out.put(targetIndex, PredicateRemapper.remap(transform,
outputRowType));
}
return out;
}
-
- private static class PredicateRemapper implements
PredicateVisitor<Predicate> {
-
- private final RowType outputRowType;
-
- private PredicateRemapper(RowType outputRowType) {
- this.outputRowType = outputRowType;
- }
-
- @Override
- public Predicate visit(LeafPredicate predicate) {
- Transform transform = predicate.transform();
- List<Object> newInputs = new ArrayList<>();
- for (Object input : transform.inputs()) {
- if (input instanceof FieldRef) {
- FieldRef ref = (FieldRef) input;
- String fieldName = ref.name();
- int newIndex = outputRowType.getFieldIndex(fieldName);
- if (newIndex < 0) {
- throw new IllegalArgumentException(
- String.format(
- "Unable to read data without column %s
when row filter enabled.",
- fieldName));
- }
- DataType type = outputRowType.getTypeAt(newIndex);
- newInputs.add(new FieldRef(newIndex, fieldName, type));
- } else {
- newInputs.add(input);
- }
- }
- return predicate.copyWithNewInputs(newInputs);
- }
-
- @Override
- public Predicate visit(CompoundPredicate predicate) {
- checkArgument(
- predicate.function() != null, "Compound row filter
function cannot be null.");
- checkArgument(
- predicate.children() != null, "Compound row filter
children cannot be null.");
- List<Predicate> remappedChildren = new ArrayList<>();
- for (Predicate child : predicate.children()) {
- checkArgument(child != null, "Compound row filter child cannot
be null.");
- Predicate remapped = child.visit(this);
- if (remapped != null) {
- remappedChildren.add(remapped);
- }
- }
- if (remappedChildren.isEmpty()) {
- throw new IllegalArgumentException("Compound row filter must
contain a predicate.");
- }
- if (remappedChildren.size() == 1) {
- return remappedChildren.get(0);
- }
- return new CompoundPredicate(predicate.function(),
remappedChildren);
- }
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
index ee054d445c..3523b1a62e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
@@ -27,6 +27,7 @@ import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateVisitor;
import org.apache.paimon.reader.DataEvolutionArray;
import org.apache.paimon.reader.DataEvolutionRow;
import org.apache.paimon.schema.SchemaManager;
@@ -215,9 +216,9 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
/**
* Per-file column pruning within a row-id-range group: drop files whose
physical columns have
- * no overlap with the query's {@code readType}. Necessary for
columnar-split DE scenarios where
- * a logical row is reconstructed from multiple files in the same row id
range — a query that
- * does not reference a file's columns has no reason to read it.
+ * no overlap with the query's {@code readType} or filter. Necessary for
columnar-split DE
+ * scenarios where a logical row is reconstructed from multiple files in
the same row id range —
+ * a query that does not reference a file's columns has no reason to read
it.
*
* <p>When every file in the group lacks a requested column (e.g. an ADD
COLUMN projection over
* a row-disjoint pre-ALTER group), one file is kept as a row-count
representative so the reader
@@ -236,6 +237,16 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
for (DataField f : readType.getFields()) {
readFieldIds.add(f.id());
}
+ if (inputFilter != null) {
+ // executeFilter may need columns absent from the output
projection. Keep their latest
+ // files too, otherwise widening the reader could see an older
value or a null.
+ Set<String> filterFields =
PredicateVisitor.collectFieldNames(inputFilter);
+ for (DataField field : schema.fields()) {
+ if (filterFields.contains(field.name())) {
+ readFieldIds.add(field.id());
+ }
+ }
+ }
List<ManifestEntry> kept = new ArrayList<>(group.size());
for (ManifestEntry entry : group) {
Set<Integer> fileIds = fileFieldIdsForEntry(entry);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index 924bb3c9dc..97762e225c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -190,12 +190,18 @@ public class FormatReadBuilder implements ReadBuilder {
protected RecordReader<InternalRow> createReader(
FormatDataSplit dataSplit, @Nullable ReadBatchSizer
readBatchSizer) throws IOException {
+ return createReader(dataSplit, readBatchSizer, readType());
+ }
+
+ protected RecordReader<InternalRow> createReader(
+ FormatDataSplit dataSplit, @Nullable ReadBatchSizer
readBatchSizer, RowType readType)
+ throws IOException {
// Skip pushing down partition filters to reader.
List<Predicate> readFilters =
excludePredicateWithFields(
PredicateBuilder.splitAnd(filter), new
HashSet<>(table.partitionKeys()));
RowType dataRowType = getRowTypeWithoutPartition(table.rowType(),
table.partitionKeys());
- RowType readRowType = getRowTypeWithoutPartition(readType(),
table.partitionKeys());
+ RowType readRowType = getRowTypeWithoutPartition(readType,
table.partitionKeys());
FormatReaderFactory readerFactory =
FileFormatDiscover.of(options)
.discover(options.formatType())
@@ -203,7 +209,7 @@ public class FormatReadBuilder implements ReadBuilder {
Pair<int[], RowType> partitionMapping =
PartitionUtils.getPartitionMapping(
- table.partitionKeys(), readType().getFields(),
table.partitionType());
+ table.partitionKeys(), readType.getFields(),
table.partitionType());
BinaryRow partition = dataSplit.partition();
FileIO fileIO =
fileIOResolver().fileIO(dataSplit.useCatalogContextFileIO());
@@ -217,7 +223,8 @@ public class FormatReadBuilder implements ReadBuilder {
partition,
readerFactory,
partitionMapping,
- readBatchSizer));
+ readBatchSizer,
+ readType));
}
return ConcatRecordReader.create(suppliers);
}
@@ -228,7 +235,8 @@ public class FormatReadBuilder implements ReadBuilder {
@Nullable BinaryRow partition,
FormatReaderFactory readerFactory,
Pair<int[], RowType> partitionMapping,
- @Nullable ReadBatchSizer readBatchSizer)
+ @Nullable ReadBatchSizer readBatchSizer,
+ RowType readType)
throws IOException {
FormatReaderContext formatReaderContext =
new FormatReaderContext(
@@ -243,7 +251,7 @@ public class FormatReadBuilder implements ReadBuilder {
reader = readerFactory.createReader(formatReaderContext);
}
return new DataFileRecordReader(
- readType(),
+ readType,
reader,
options.scanIgnoreCorruptFile(),
options.scanIgnoreLostFile(),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
index 18e3ce3505..51f3d2b1e4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRead.java
@@ -22,11 +22,11 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateProjectionConverter;
import org.apache.paimon.reader.LimitRecordReader;
import org.apache.paimon.reader.ReadBatchSizer;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.source.ReadTransform;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.RowType;
@@ -34,7 +34,7 @@ import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
import java.io.IOException;
-import java.util.Optional;
+import java.util.Collections;
/** A {@link TableRead} implementation for {@link FormatTable}. */
public class FormatTableRead implements TableRead {
@@ -55,7 +55,7 @@ public class FormatTableRead implements TableRead {
Predicate predicate,
Integer limit) {
this.tableRowType = tableRowType;
- this.readType = readType;
+ this.readType = readType == null ? tableRowType : readType;
this.read = read;
this.predicate = predicate;
this.limit = limit;
@@ -89,30 +89,16 @@ public class FormatTableRead implements TableRead {
// Capture the binding per TableRead so lazy file suppliers cannot
observe another read's
// sizer.
ReadBatchSizer sizer = this.readBatchSizer;
- RecordReader<InternalRow> reader = read.createReader(dataSplit, sizer);
- if (executeFilter) {
- reader = executeFilter(reader);
- }
+ ReadTransform transform =
+ ReadTransform.create(
+ tableRowType,
+ readType,
+ predicate,
+ executeFilter,
+ null,
+ Collections.emptySet());
+ RecordReader<InternalRow> reader =
+ transform.apply(read.createReader(dataSplit, sizer,
transform.readType()));
return LimitRecordReader.limit(reader, limit);
}
-
- private RecordReader<InternalRow> executeFilter(RecordReader<InternalRow>
reader) {
- if (predicate == null) {
- return reader;
- }
-
- Predicate predicate = this.predicate;
- if (readType != null) {
- int[] projection =
tableRowType.getFieldIndices(readType.getFieldNames());
- Optional<Predicate> optional =
-
predicate.visit(PredicateProjectionConverter.fromProjection(projection));
- if (!optional.isPresent()) {
- return reader;
- }
- predicate = optional.get();
- }
-
- Predicate finalFilter = predicate;
- return reader.filter(finalFilter::test);
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java
index 7b14dd1cac..4c46ddd4cb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java
@@ -23,24 +23,14 @@ import org.apache.paimon.catalog.TableQueryAuthResult;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateProjectionConverter;
-import org.apache.paimon.predicate.PredicateVisitor;
-import org.apache.paimon.predicate.Transform;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.ProjectedRow;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
import java.util.Set;
/** A {@link InnerTableRead} for data table. */
@@ -55,9 +45,6 @@ public abstract class AbstractDataTableRead implements
InnerTableRead {
// as read-level TopN already does (see ReadBuilderImpl)
private final boolean queryAuthEnabled;
- // the auth-widened read type currently applied, or null when the plain
read type is
- @Nullable private RowType appliedReadType;
-
// blob-view columns that only resolve through the dedicated blob-view
read path
private final Set<String> resolvedBlobViewFields;
@@ -113,7 +100,6 @@ public abstract class AbstractDataTableRead implements
InnerTableRead {
@Override
public final InnerTableRead withReadType(RowType readType) {
this.readType = readType;
- this.appliedReadType = null;
applyReadType(readType);
return this;
}
@@ -147,173 +133,26 @@ public abstract class AbstractDataTableRead implements
InnerTableRead {
protected final RecordReader<InternalRow> createDataReader(
Split split, @Nullable TableQueryAuthResult authResult) throws
IOException {
- // A TableRead can be reused for multiple splits. Authentication may
have expanded an
- // explicitly configured physical projection for the previous split,
so restore it before
- // applying the current split's authorization dependencies. Without an
explicit projection,
- // the underlying reader must retain its own default read type.
- if (readType != null) {
- applyReadType(readType);
- appliedReadType = null;
- }
- RecordReader<InternalRow> reader;
- if (authResult == null) {
- reader = backProject(reader(split));
- } else {
- reader = authedReader(split, authResult);
- }
- if (executeFilter) {
- reader = executeFilter(reader);
- }
-
- return reader;
- }
-
- private RecordReader<InternalRow> authedReader(Split split,
TableQueryAuthResult authResult)
- throws IOException {
- List<String> readFields = currentReadType().getFieldNames();
- // masked filter columns are read and masked like rule fields, then
evaluated post-mask
- Set<String> maskedFilterFields =
- maskedFilterFields(authResult.extractColumnMasking().keySet());
- Set<String> ruleFields = authResult.authFields(readFields, predicate);
- RowType widened = widenedReadType(authResult, ruleFields);
- if (widened != null && !widened.equals(appliedReadType)) {
- applyReadType(widened);
- appliedReadType = widened;
- }
- // the split read emits appliedReadType; rules are remapped against it
by name
- RowType outputType = appliedReadType != null ? appliedReadType :
currentReadType();
- // masks apply only to columns readable from the query: the ones it
projects plus the
- // ones the rules pulled in; a mask on anything else is inert
- Map<String, Transform> masking = authResult.extractColumnMasking();
- Map<String, Transform> selectedColumnMasking = Collections.emptyMap();
- if (!masking.isEmpty()) {
- Set<String> activeFields = new HashSet<>(readFields);
- activeFields.addAll(ruleFields);
- selectedColumnMasking = new HashMap<>();
- for (Map.Entry<String, Transform> mask : masking.entrySet()) {
- if (activeFields.contains(mask.getKey())) {
- selectedColumnMasking.put(mask.getKey(), mask.getValue());
- }
- }
- }
- RecordReader<InternalRow> reader =
- authResult.doAuth(
- reader(split),
- outputType,
- authResult.extractPredicate(),
- selectedColumnMasking);
- reader = filterMaskedConjuncts(reader, outputType, maskedFilterFields);
- return backProject(reader);
- }
-
- private Set<String> maskedFilterFields(Set<String> maskTargets) {
- if (predicate == null || maskTargets.isEmpty()) {
- return Collections.emptySet();
- }
- Set<String> fields = new
HashSet<>(PredicateVisitor.collectFieldNames(predicate));
- fields.retainAll(maskTargets);
- return fields;
- }
-
- /**
- * Evaluates the filter conjuncts on masked columns, on the masked output:
they are never pushed
- * down, and engines do not re-evaluate the conjuncts they consumed.
- */
- private RecordReader<InternalRow> filterMaskedConjuncts(
- RecordReader<InternalRow> reader, RowType outputType, Set<String>
maskedFilterFields) {
- if (maskedFilterFields.isEmpty()) {
- return reader;
- }
- Predicate maskedPart = TableQueryAuthResult.retainFields(predicate,
maskedFilterFields);
- if (maskedPart == null) {
- return reader;
- }
- // by name against the emitted schema: it may carry system fields the
table schema lacks
- Predicate filter;
- try {
- filter = TableQueryAuthResult.remapPredicate(maskedPart,
outputType);
- } catch (RuntimeException e) {
- throw new IllegalStateException(
- "Filter on masked columns "
- + maskedFilterFields
- + " cannot be evaluated on read schema "
- + outputType.getFieldNames(),
- e);
- }
- return reader.filter(filter::test);
- }
-
- /** Project auth-widened rows back to the read type the query asked for. */
- private RecordReader<InternalRow> backProject(RecordReader<InternalRow>
reader) {
- if (appliedReadType == null) {
- return reader;
- }
- ProjectedRow backRow =
- ProjectedRow.from(
-
appliedReadType.projectIndexes(currentReadType().getFieldNames()));
- return reader.transform(backRow::replaceRow);
- }
-
- /**
- * The read type widened with the unprojected columns the auth rules read,
or null when the
- * projection already covers them. Projected fields are kept as-is to
preserve nested pruning.
- */
- @Nullable
- private RowType widenedReadType(TableQueryAuthResult authResult,
Set<String> ruleFields) {
- RowType tableType = schema.logicalRowType();
- RowType readType = currentReadType();
- Set<String> maskTargets = authResult.extractColumnMasking().keySet();
- for (String name : readType.getFieldNames()) {
- if (!ruleFields.contains(name) && !maskTargets.contains(name)) {
- continue;
- }
- if (!tableType.containsField(name)) {
- continue;
- }
- // rules must not touch a nested-pruned column (partial value)
- DataField tableField = tableType.getField(name);
- if (!readType.getField(name).type().equals(tableField.type())) {
- throw new IllegalStateException(
- String.format(
- "Query auth rules involve column '%s', which
the query "
- + "projects with a pruned type %s
instead of its "
- + "table type %s; cannot apply the
rules to a "
- + "partial column.",
- name, readType.getField(name).type(),
tableField.type()));
- }
- }
- for (String name : ruleFields) {
- if (resolvedBlobViewFields.contains(name) &&
!readType.containsField(name)) {
- // auth-added columns bypass blob-view resolution
- throw new IllegalStateException(
- String.format(
- "Query auth rules read blob-view column '%s',
which the query "
- + "does not project; such columns
cannot be resolved. "
- + "Project the column or adjust the
rule.",
- name));
+ if (authResult == null && !(executeFilter && predicate != null)) {
+ // Restore an explicit projection after a previous split needed
authorization columns.
+ // Without a projection, preserve the underlying reader's default
read type.
+ if (readType != null) {
+ applyReadType(readType);
}
- }
- return TableQueryAuthResult.appendMissingFields(tableType, readType,
ruleFields);
- }
-
- private RecordReader<InternalRow> executeFilter(RecordReader<InternalRow>
reader) {
- if (predicate == null) {
- return reader;
- }
-
- Predicate predicate = this.predicate;
- if (readType != null) {
- int[] projection =
schema.logicalRowType().getFieldIndices(readType.getFieldNames());
- Optional<Predicate> optional =
-
predicate.visit(PredicateProjectionConverter.fromProjection(projection));
- if (!optional.isPresent()) {
- return reader;
- }
- predicate = optional.get();
- }
-
- Predicate finalFilter = predicate;
- return reader.filter(finalFilter::test);
+ return reader(split);
+ }
+ ReadTransform transform =
+ ReadTransform.create(
+ schema.logicalRowType(),
+ currentReadType(),
+ predicate,
+ executeFilter,
+ authResult,
+ resolvedBlobViewFields);
+ if (readType != null ||
!transform.readType().equals(currentReadType())) {
+ applyReadType(transform.readType());
+ }
+ return transform.apply(reader(split));
}
/** Split with auth context. */
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
index 0cdef986fe..8be73caff6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
@@ -30,6 +30,7 @@ import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateRemapper;
import org.apache.paimon.predicate.PredicateVisitor;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
@@ -62,14 +63,17 @@ import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RowRangeIndex;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
+import org.apache.paimon.utils.TypeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -167,7 +171,7 @@ abstract class AbstractDataTableScan implements
DataTableScan {
// Remap field-id FieldRefs to positional indices by name (as
doAuth does on read), so
// pruning stays correct across schema evolution.
Predicate remappedAuth =
- TableQueryAuthResult.remapPredicate(authPredicate,
schema.logicalRowType());
+ PredicateRemapper.remap(authPredicate,
schema.logicalRowType());
if (remappedAuth != null) {
Pair<Optional<PartitionPredicate>, List<Predicate>> split =
PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
@@ -279,6 +283,14 @@ abstract class AbstractDataTableScan implements
DataTableScan {
return null;
}
List<String> select = readType == null ? null :
readType.getFieldNames();
+ if (select != null && (userFilter != null ||
!partitionFilterFields.isEmpty())) {
+ // Authorize query operands before pruning or reading their
values. Dependencies of
+ // trusted row filters and masks are supplied by the catalog and
are not user selects.
+ Set<String> fields = new LinkedHashSet<>(select);
+ fields.addAll(PredicateVisitor.collectFieldNames(userFilter));
+ fields.addAll(partitionFilterFields);
+ select = new ArrayList<>(fields);
+ }
TableQueryAuthResult result = queryAuth.auth(select);
if (result != null && result.hasRules()) {
// re-validated every plan, so a schema change under a live scan
fails closed
@@ -320,11 +332,23 @@ abstract class AbstractDataTableScan implements
DataTableScan {
/** The partition columns a predicate references, or all of them when it
cannot be read. */
private Set<String> partitionPredicateFields(PartitionPredicate
partitionPredicate) {
+ if (partitionPredicate == PartitionPredicate.ALWAYS_TRUE
+ || partitionPredicate == PartitionPredicate.ALWAYS_FALSE) {
+ return Collections.emptySet();
+ }
if (partitionPredicate instanceof
PartitionPredicate.DefaultPartitionPredicate) {
return PredicateVisitor.collectFieldNames(
((PartitionPredicate.DefaultPartitionPredicate)
partitionPredicate)
.predicate());
}
+ if (partitionPredicate instanceof
PartitionPredicate.AndPartitionPredicate) {
+ Set<String> fields = new HashSet<>();
+ for (PartitionPredicate child :
+ ((PartitionPredicate.AndPartitionPredicate)
partitionPredicate).predicates()) {
+ fields.addAll(partitionPredicateFields(child));
+ }
+ return fields;
+ }
return new HashSet<>(schema.partitionKeys());
}
@@ -383,34 +407,24 @@ abstract class AbstractDataTableScan implements
DataTableScan {
}
/**
- * Push the auth-widened read type to the snapshot reader before planning,
so file-level column
- * pruning keeps the files of the columns the rules read.
+ * Push the read type expanded for filters and auth rules to the snapshot
reader before
+ * planning, so file-level column pruning keeps their dependencies.
*/
private void applyAuthReadType(@Nullable TableQueryAuthResult
queryAuthResult) {
if (readType == null) {
return;
}
- RowType desired = readType;
- if (queryAuthResult != null && queryAuthResult.hasRules()) {
- // post-mask conjuncts are evaluated at read time; their columns
must survive planning
- RowType widened =
- TableQueryAuthResult.appendMissingFields(
- schema.logicalRowType(),
- readType,
-
queryAuthResult.authFields(readType.getFieldNames(), userFilter));
- if (widened != null) {
- desired = widened;
- }
- }
- // never narrow within this scan's lifetime: readers fix their schema
on first use
- RowType widenedToApplied =
- TableQueryAuthResult.appendMissingFields(
+ RowType desired =
+ TypeUtils.withMissingFields(
+ schema.logicalRowType(),
+ readType,
+ ReadTransform.requiredFields(readType, userFilter,
queryAuthResult));
+ // Never narrow within this scan's lifetime: readers may retain their
physical schema.
+ desired =
+ TypeUtils.withMissingFields(
appliedScanReadType,
desired,
new HashSet<>(appliedScanReadType.getFieldNames()));
- if (widenedToApplied != null) {
- desired = widenedToApplied;
- }
if (!desired.equals(appliedScanReadType)) {
snapshotReader.withReadType(desired);
appliedScanReadType = desired;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadTransform.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadTransform.java
new file mode 100644
index 0000000000..33f2722cc9
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadTransform.java
@@ -0,0 +1,178 @@
+/*
+ * 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.table.source;
+
+import org.apache.paimon.catalog.TableQueryAuthResult;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateRemapper;
+import org.apache.paimon.predicate.PredicateVisitor;
+import org.apache.paimon.predicate.Transform;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.NestedProjectedRow;
+import org.apache.paimon.utils.TypeUtils;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * The physical read type and ordered transformations for one reader. Create a
new instance for each
+ * split so authorization rules and query settings cannot leak between readers.
+ */
+public final class ReadTransform {
+
+ private final RowType readType;
+ private final RowType outputType;
+ @Nullable private final TableQueryAuthResult authResult;
+ private final Map<String, Transform> masking;
+ @Nullable private final Predicate filter;
+
+ private ReadTransform(
+ RowType readType,
+ RowType outputType,
+ @Nullable TableQueryAuthResult authResult,
+ Map<String, Transform> masking,
+ @Nullable Predicate filter) {
+ this.readType = readType;
+ this.outputType = outputType;
+ this.authResult = authResult;
+ this.masking = masking;
+ this.filter = filter;
+ }
+
+ public static ReadTransform create(
+ RowType tableType,
+ RowType outputType,
+ @Nullable Predicate queryFilter,
+ boolean executeFilter,
+ @Nullable TableQueryAuthResult authResult,
+ Set<String> resolvedBlobViewFields) {
+ Map<String, Transform> masks =
+ authResult == null ? Collections.emptyMap() :
authResult.extractColumnMasking();
+ // Without executeFilter, engines still rely on us to evaluate
conjuncts on masked columns.
+ Predicate filter =
+ executeFilter ? queryFilter : maskedQueryFilter(queryFilter,
masks.keySet());
+ RowType filterType =
+ executeFilter && queryFilter != null
+ ? withFilterFields(tableType, outputType, queryFilter)
+ : outputType;
+ Set<String> required = requiredFields(outputType, filter, authResult);
+ if (authResult != null) {
+ authResult.validateReadType(tableType, filterType, required,
resolvedBlobViewFields);
+ }
+ RowType readType = TypeUtils.withMissingFields(tableType, filterType,
required);
+
+ Map<String, Transform> selectedMasks = new HashMap<>();
+ Set<String> active = new HashSet<>(filterType.getFieldNames());
+ active.addAll(required);
+ for (Map.Entry<String, Transform> mask : masks.entrySet()) {
+ if (active.contains(mask.getKey())) {
+ selectedMasks.put(mask.getKey(), mask.getValue());
+ }
+ }
+ if (filter != null) {
+ try {
+ filter = PredicateRemapper.remap(filter, readType);
+ } catch (RuntimeException e) {
+ if (executeFilter) {
+ throw e;
+ }
+ throw new IllegalStateException(
+ "Filter on masked columns "
+ + masks.keySet()
+ + " cannot be evaluated on read schema "
+ + readType.getFieldNames(),
+ e);
+ }
+ }
+ return new ReadTransform(readType, outputType, authResult,
selectedMasks, filter);
+ }
+
+ /**
+ * Columns needed in addition to the output projection. Planning passes
the full query filter so
+ * its files survive pruning; readers pass the query predicate they will
actually execute.
+ */
+ public static Set<String> requiredFields(
+ RowType outputType,
+ @Nullable Predicate queryFilter,
+ @Nullable TableQueryAuthResult authResult) {
+ Set<String> fields = new
HashSet<>(PredicateVisitor.collectFieldNames(queryFilter));
+ if (authResult != null) {
+ Set<String> visible = new HashSet<>(outputType.getFieldNames());
+ visible.addAll(fields);
+ fields.addAll(authResult.authFields(new ArrayList<>(visible),
queryFilter));
+ }
+ return fields;
+ }
+
+ public RowType readType() {
+ return readType;
+ }
+
+ /** Apply authorization, masking, the query filter, and the output
projection, in that order. */
+ public RecordReader<InternalRow> apply(RecordReader<InternalRow> reader) {
+ if (authResult != null) {
+ reader = authResult.doAuth(reader, readType,
authResult.extractPredicate(), masking);
+ }
+ if (filter != null) {
+ reader = reader.filter(filter::test);
+ }
+ NestedProjectedRow projection = NestedProjectedRow.create(readType,
outputType);
+ return projection == null ? reader :
reader.transform(projection::replaceRow);
+ }
+
+ @Nullable
+ private static Predicate maskedQueryFilter(
+ @Nullable Predicate filter, Set<String> maskTargets) {
+ return filter == null || maskTargets.isEmpty()
+ ? null
+ : TableQueryAuthResult.retainFields(filter, maskTargets);
+ }
+
+ private static RowType withFilterFields(
+ RowType tableType, RowType readType, Predicate predicate) {
+ Set<String> fields = PredicateVisitor.collectFieldNames(predicate);
+ RowType widened = TypeUtils.withMissingFields(tableType, readType,
fields);
+ List<DataField> fullFields = new ArrayList<>(widened.getFields());
+ for (int i = 0; i < fullFields.size(); i++) {
+ DataField field = fullFields.get(i);
+ if (fields.contains(field.name()) &&
tableType.containsField(field.name())) {
+ // Filter operands use their full type; nested pruning is
restored after filtering.
+ fullFields.set(i, tableType.getField(field.name()));
+ }
+ }
+ checkArgument(
+ widened.getFieldNames().containsAll(fields),
+ "Cannot execute filter on fields %s with read type %s.",
+ fields,
+ widened);
+ return widened.copy(fullFields);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java
index 930a924ae6..91e0b7d014 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/TableRead.java
@@ -43,6 +43,18 @@ public interface TableRead {
/** Set {@link MetricRegistry} to table read. */
TableRead withMetricRegistry(MetricRegistry registry);
+ /**
+ * Enable row-level evaluation of the complete configured filter for
subsequently created
+ * readers. Without this call, filter pushdown may only prune files or row
groups and can return
+ * rows that do not satisfy the filter.
+ *
+ * <p>Fields referenced by the filter are read even if they are absent
from the requested read
+ * type. The filter is evaluated before the final output projection, so
the returned rows retain
+ * the requested fields and their order. For tables with query
authorization, the filter is
+ * evaluated after authorization and column masking.
+ *
+ * @return this read
+ */
TableRead executeFilter();
TableRead withIOManager(IOManager ioManager);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 1cf891dbd3..05f62688eb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -51,6 +51,7 @@ import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.operation.FileStoreWrite;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.CastTransform;
import org.apache.paimon.predicate.ConcatTransform;
@@ -126,6 +127,8 @@ import org.apache.commons.io.FileUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.Mockito;
import java.io.File;
@@ -3097,6 +3100,84 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
table.newReadBuilder().withProjection(new int[] {1}).newScan().plan();
}
+ @ParameterizedTest
+ @CsvSource({"false,false", "true,false", "false,true", "true,true"})
+ void testTableAuthIncludesUnprojectedFilterFields(boolean streaming,
boolean partitionFilter)
+ throws Exception {
+ Identifier identifier = Identifier.create("test_table_db",
"auth_filter_columns");
+ Table table =
+ createMaskingAuthTable(
+ identifier,
+ Arrays.asList(
+ new DataField(0, "public_id", DataTypes.INT()),
+ new DataField(1, "secret_score",
DataTypes.INT())),
+ partitionFilter ? singletonList("secret_score") :
Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap());
+ commitRows(table, GenericRow.of(1, 10), GenericRow.of(2, 20),
GenericRow.of(3, 30));
+ authTableColumns(identifier, singletonList("public_id"));
+
+ assertThatThrownBy(
+ () -> table.newReadBuilder().withProjection(new int[]
{1}).newScan().plan())
+ .hasMessageContaining("has no permission");
+
+ ReadBuilder readBuilder = table.newReadBuilder().withProjection(new
int[] {0});
+ if (partitionFilter) {
+ readBuilder.withPartitionFilter(singletonMap("secret_score",
"20"));
+ } else {
+ readBuilder.withFilter(new
PredicateBuilder(table.rowType()).equal(1, 20));
+ }
+ TableScan scan = streaming ? readBuilder.newStreamScan() :
readBuilder.newScan();
+ TableRead read = readBuilder.newRead().executeFilter();
+ assertThatThrownBy(
+ () -> {
+ try (RecordReader<InternalRow> reader =
+ read.createReader(scan.plan().splits())) {
+ reader.forEachRemaining(row -> {});
+ }
+ })
+ .hasMessageContaining("has no permission");
+
+ // Granting the filter operand allows the same query without exposing
the extra column.
+ authTableColumns(identifier, Arrays.asList("public_id",
"secret_score"));
+ List<InternalRow> rows =
+ collectRows(
+ read.createReader(scan.plan().splits()),
+ table.rowType().project("public_id"));
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getFieldCount()).isEqualTo(1);
+ assertThat(rows.get(0).getInt(0)).isEqualTo(2);
+ }
+
+ @Test
+ void testTableAuthWithCombinedPartitionFilter() throws Exception {
+ Identifier identifier =
+ Identifier.create("test_table_db",
"auth_combined_partition_filter");
+ Table table =
+ createMaskingAuthTable(
+ identifier,
+ stringFields("p1", "p2", "v"),
+ Arrays.asList("p1", "p2"),
+ Collections.emptyList(),
+ Collections.emptyMap());
+ writeStringRows(table, new String[] {"x", "a", "v1"}, new String[]
{"y", "b", "v2"});
+ authTableColumns(identifier, Arrays.asList("p1", "v"));
+
+ RowType partitionType = table.rowType().project(table.partitionKeys());
+ Predicate onP1 = new PredicateBuilder(partitionType).equal(0,
BinaryString.fromString("x"));
+ ReadBuilder readBuilder =
+ table.newReadBuilder()
+ .withProjection(new int[] {2})
+ .withPartitionFilter(
+ PartitionPredicate.and(
+ Arrays.asList(
+
PartitionPredicate.fromPredicate(
+ partitionType, onP1),
+
PartitionPredicate.ALWAYS_TRUE)));
+ assertThat(batchRead(table, readBuilder.newScan().plan().splits(),
readBuilder))
+ .containsExactly("+I[v1]");
+ }
+
@Test
void testSnapshotMethods() throws Exception {
Identifier identifier = Identifier.create("test_table_db",
"snapshots_table");
@@ -4076,6 +4157,9 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
new String[] {"jane", "roe", "ignored", "o2"});
maskDisplayWithFullName(identifier);
+ // Trusted masking inputs do not require the caller to have column
access.
+ authTableColumns(identifier, singletonList("display"));
+
ReadBuilder readBuilder = table.newReadBuilder().withProjection(new
int[] {2});
List<Split> splits = readBuilder.newScan().plan().splits();
List<InternalRow> rows =
@@ -4145,6 +4229,9 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
setRowFilter(identifier, Collections.singletonList(displayFilter));
maskDisplayWithFullName(identifier);
+ // Trusted row-filter inputs do not require the caller to have column
access either.
+ authTableColumns(identifier, singletonList("other"));
+
ReadBuilder readBuilder = table.newReadBuilder().withProjection(new
int[] {3});
List<Split> splits = readBuilder.newScan().plan().splits();
List<InternalRow> rows =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
index 0ee799c032..dd6f531f8c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
@@ -614,7 +614,7 @@ public class AppendOnlySimpleTableTest extends
SimpleTableTestBase {
assertThat(getResult(read, splits, binaryRow(2), 0, toString))
.hasSameElementsAs(Arrays.asList("201|binary", "201|binary"));
- // projection contains unknown index or
+ // OR includes a field outside the output projection.
read =
table.newRead()
.withFilter(
@@ -622,10 +622,9 @@ public class AppendOnlySimpleTableTest extends
SimpleTableTestBase {
.withProjection(new int[] {3, 2})
.executeFilter();
assertThat(getResult(read, splits, binaryRow(2), 0, toString))
- .hasSameElementsAs(
- Arrays.asList("200|binary", "201|binary",
"202|binary", "201|binary"));
+ .hasSameElementsAs(Arrays.asList("201|binary", "201|binary"));
- // projection contains unknown index and
+ // AND must evaluate the unprojected partition field too.
read =
table.newRead()
.withFilter(
@@ -633,8 +632,7 @@ public class AppendOnlySimpleTableTest extends
SimpleTableTestBase {
.withProjection(new int[] {3, 2})
.executeFilter();
assertThat(getResult(read, splits, binaryRow(1), 0,
toString)).isEmpty();
- assertThat(getResult(read, splits, binaryRow(2), 0, toString))
- .hasSameElementsAs(Arrays.asList("201|binary", "201|binary"));
+ assertThat(getResult(read, splits, binaryRow(2), 0,
toString)).isEmpty();
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
index a45a508057..b3cfa4965d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java
@@ -89,9 +89,8 @@ import static org.assertj.core.api.Assertions.assertThat;
* down is an optimization and must never change the result. A row count alone
can not say that, it
* passes just as well when nothing is pushed down at all.
*
- * <p>A filter on a column that is not part of the read type has no {@link
#query} counterpart:
- * {@link TableRead#executeFilter()} can not project such a predicate onto the
read row and silently
- * keeps every row, so only the split level assertion says anything.
+ * <p>Filters on unprojected columns are also checked through {@link
TableRead#executeFilter()},
+ * which reads the filter operands before restoring the requested output
projection.
*
* <p>Filter values are always picked inside the min/max range of the column,
otherwise the group
* level stats pruning in {@link
org.apache.paimon.operation.DataEvolutionFileStoreScan} would drop
@@ -324,15 +323,38 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
assertThat(rows).anyMatch(row -> row.getInt(0) == 50);
}
+ @Test
+ public void testExecuteFilterWithUnprojectedOverwrittenColumn() throws
Exception {
+ FileStoreTable table = createTable("execute_filter_projection",
Collections.emptyMap());
+ writeThenOverwriteF1(table, ROW_COUNT);
+ FileStoreTable latest = getTable(identifier(table.name()));
+ RowType outputType = rowType().project("f0");
+ ReadBuilder readBuilder =
+
latest.newReadBuilder().withReadType(outputType).withFilter(equalF1(c1(50)));
+ List<InternalRow> rows =
+ collect(
+ readBuilder.newRead().executeFilter(),
+ readBuilder.newScan().plan(),
+ outputType);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getInt(0)).isEqualTo(50);
+
+ readBuilder =
latest.newReadBuilder().withReadType(outputType).withFilter(equalF1(f1(50)));
+ assertThat(
+ collect(
+ readBuilder.newRead().executeFilter(),
+ readBuilder.newScan().plan(),
+ outputType))
+ .isEmpty();
+ }
+
@Test
public void testFileIndexIsGivenUpForAColumnOutsideTheReadType() throws
Exception {
FileStoreTable table = createTable("projection_index",
bloomOptions("f1", "1 B"));
writeAllColumns(table, ROW_COUNT);
- // the index of this file does prove that no row of it matches, but a
file only owns the
- // columns of the read type: the file that owns f1 can be pruned out
of the split, see
- // testProjectionPruningAwayTheWinnerOfTheFilterColumn, so the push
down gives the column
- // up rather than trust whichever file is left holding it
+ // The split reader only evaluates filters over its read type. Without
executeFilter,
+ // f1 remains outside that type even though planning retains its files.
assertThat(readWithFilter(table, equalF1(MISSING_F1),
rowType().project("f0")))
.hasSize(ROW_COUNT);
@@ -356,14 +378,12 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
}
@Test
- public void testProjectionPruningAwayTheWinnerOfTheFilterColumn() throws
Exception {
+ public void testUnprojectedOverwrittenFilterColumnWithoutExecution()
throws Exception {
FileStoreTable table = createTable("pruned_winner",
Collections.emptyMap());
writeThenOverwriteF1(table, ROW_COUNT);
- // f1 was rewritten as c* by a second file, and projecting f0 prunes
that file out of the
- // split, see DataEvolutionFileStoreScan#pruneByReadType. The old file
is left holding a*
- // and a bloom index that knows nothing about c*, so nothing about it
may be used to prove
- // that a row does not match: the row does match, through the file
that is not there.
+ // f1 was rewritten as c* by a second file. Without executeFilter the
reader still only
+ // reads f0, so an unprojected filter must not reject the matching row.
List<InternalRow> rows = readWithFilter(table, equalF1(c1(50)),
rowType().project("f0"));
assertThat(rows).anyMatch(row -> row.getInt(0) == 50);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 6a7a722e85..12339f35b9 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -585,6 +585,32 @@ public class PrimaryKeySimpleTableTest extends
SimpleTableTestBase {
"2|22|202|binary|varbinary|mapKey:mapVal|multiset"));
}
+ @Test
+ public void testExecuteFilterWithUnprojectedValue() throws Exception {
+ writeData();
+ FileStoreTable table = createFileStoreTable();
+ List<Split> splits =
toSplits(table.newSnapshotReader().read().dataSplits());
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ TableRead read =
+ table.newRead()
+ .withReadType(table.rowType().project(new int[] {1,
0}))
+ .withFilter(builder.equal(2, 20001L))
+ .executeFilter();
+ Function<InternalRow, String> toString =
+ row -> {
+ assertThat(row.getFieldCount()).isEqualTo(2);
+ return row.getInt(0) + "|" + row.getInt(1);
+ };
+ assertThat(getResult(read, splits, toString)).containsExactly("21|2");
+ // The old value of the same primary key must not survive merge and
filtering.
+ read =
+ table.newRead()
+ .withReadType(table.rowType().project(new int[] {1,
0}))
+ .withFilter(builder.equal(2, 201L))
+ .executeFilter();
+ assertThat(getResult(read, splits, toString)).isEmpty();
+ }
+
@Test
public void testBranchBatchReadWrite() throws Exception {
FileStoreTable table = createFileStoreTable();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableReadTest.java
new file mode 100644
index 0000000000..dccbdc0ac1
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableReadTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.table.format;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests exact filtering with projected columns in {@link FormatTableRead}. */
+class FormatTableReadTest {
+
+ @TempDir Path tempDir;
+
+ @Test
+ void testExecuteFilterWithUnprojectedFields() throws Exception {
+ RowType type = RowType.of(DataTypes.INT(), DataTypes.INT(),
DataTypes.STRING());
+ Files.write(
+ tempDir.resolve("data.csv"),
+ Arrays.asList("1,1,first", "1,2,second", "2,2,third",
"2,3,last"),
+ StandardCharsets.UTF_8);
+ FormatTable table =
+ FormatTable.builder()
+ .fileIO(LocalFileIO.create())
+ .identifier(Identifier.create("test_db", "test_table"))
+ .rowType(type)
+ .partitionKeys(Collections.emptyList())
+ .location(tempDir.toString())
+ .format(FormatTable.Format.CSV)
+ .options(Collections.singletonMap("file.format",
"csv"))
+ .build();
+ PredicateBuilder predicateBuilder = new PredicateBuilder(type);
+ List<Predicate> filters =
+ Arrays.asList(
+ predicateBuilder.equal(1, 2),
+ PredicateBuilder.and(
+ predicateBuilder.equal(0, 1),
predicateBuilder.equal(1, 2)),
+ PredicateBuilder.or(
+ predicateBuilder.equal(0, 1),
predicateBuilder.equal(1, 2)));
+ List<List<String>> expected =
+ Arrays.asList(
+ Arrays.asList("second:1", "third:2"),
+ Collections.singletonList("second:1"),
+ Arrays.asList("first:1", "second:1", "third:2"));
+ TableScan.Plan plan = table.newReadBuilder().newScan().plan();
+ for (int i = 0; i < filters.size(); i++) {
+ ReadBuilder builder =
+ table.newReadBuilder()
+ .withReadType(type.project(new int[] {2, 0}))
+ .withFilter(filters.get(i));
+ TableRead unfiltered = builder.newRead();
+ TableRead filtered = builder.newRead().executeFilter();
+ assertThat(readRows(filtered,
plan)).containsExactlyElementsOf(expected.get(i));
+ assertThat(readRows(filtered,
plan)).containsExactlyElementsOf(expected.get(i));
+ // Expanding one TableRead must not change another read sharing
the builder.
+ assertThat(readRows(unfiltered, plan))
+ .containsExactly("first:1", "second:1", "third:2",
"last:2");
+ }
+ }
+
+ private static List<String> readRows(TableRead read, TableScan.Plan plan)
throws Exception {
+ List<String> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader = read.createReader(plan)) {
+ reader.forEachRemaining(
+ row -> {
+ assertThat(row.getFieldCount()).isEqualTo(2);
+ result.add(row.getString(0) + ":" + row.getInt(1));
+ });
+ }
+ return result;
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java
index fa1934c4fe..44417bebd4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/AbstractDataTableReadTest.java
@@ -19,31 +19,375 @@
package org.apache.paimon.table.source;
import org.apache.paimon.catalog.TableQueryAuthResult;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.FieldTransform;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.Transform;
import org.apache.paimon.predicate.UpperTransform;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IteratorRecordReader;
import org.apache.paimon.utils.JsonSerdeUtil;
+import org.apache.paimon.utils.NestedProjectedRow;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
-/** Tests query-authorization projection expansion in {@link
AbstractDataTableRead}. */
+/**
+ * Tests filtering and query-authorization projection expansion in {@link
AbstractDataTableRead}.
+ */
class AbstractDataTableReadTest {
+ @Test
+ void testExecuteFilterWithUnprojectedFields() throws IOException {
+ RowType type = RowType.of(DataTypes.INT(), DataTypes.INT(),
DataTypes.STRING());
+ TestingDataTableRead read =
+ new TestingDataTableRead(
+ schema(type),
+ GenericRow.of(1, 1, BinaryString.fromString("first")),
+ GenericRow.of(1, 2, BinaryString.fromString("second")),
+ GenericRow.of(2, 2, BinaryString.fromString("third")),
+ GenericRow.of(2, null,
BinaryString.fromString("null")),
+ GenericRow.of(2, 3, BinaryString.fromString("last")));
+ PredicateBuilder builder = new PredicateBuilder(type);
+ RowType outputType = type.project(new int[] {2, 0});
+ read.withReadType(outputType);
+ read.executeFilter();
+
+ read.withFilter(builder.equal(1, 2));
+ assertThat(readRows(read, outputType)).containsExactly("second:1",
"third:2");
+ // Reuse the same read across splits, then change its filter.
+ assertThat(readRows(read, outputType)).containsExactly("second:1",
"third:2");
+ read.withFilter(PredicateBuilder.and(builder.equal(0, 1),
builder.equal(1, 2)));
+ assertThat(readRows(read, outputType)).containsExactly("second:1");
+ read.withFilter(PredicateBuilder.or(builder.equal(0, 1),
builder.equal(1, 2)));
+ assertThat(readRows(read, outputType)).containsExactly("first:1",
"second:1", "third:2");
+ read.withFilter(builder.isNull(1));
+ assertThat(readRows(read, outputType)).containsExactly("null:2");
+
+ read.withReadType(RowType.of());
+ read.withFilter(builder.equal(1, 2));
+ List<Integer> arities = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
read.createReader(mock(Split.class))) {
+ reader.forEachRemaining(row -> arities.add(row.getFieldCount()));
+ }
+ assertThat(arities).containsExactly(0, 0);
+ }
+
+ @Test
+ void testExecuteFilterOnUnprojectedMaskedField() throws IOException {
+ RowType type = RowType.of(DataTypes.INT(), DataTypes.STRING(),
DataTypes.STRING());
+ TestingDataTableRead read =
+ new TestingDataTableRead(
+ schema(type),
+ GenericRow.of(
+ 1,
+ BinaryString.fromString("wrong"),
+ BinaryString.fromString("match")),
+ GenericRow.of(
+ 2,
+ BinaryString.fromString("MATCH"),
+ BinaryString.fromString("no")));
+ read.withReadType(type.project(new int[] {0}));
+ read.withFilter(new PredicateBuilder(type).equal(1,
BinaryString.fromString("MATCH")));
+ read.executeFilter();
+ TableQueryAuthResult auth =
+ new TableQueryAuthResult(
+ null,
+ Collections.singletonMap(
+ type.getFieldNames().get(1),
+ JsonSerdeUtil.toFlatJson(
+ new UpperTransform(
+ Collections.singletonList(
+ new FieldRef(
+ 2,
+
type.getFieldNames().get(2),
+
DataTypes.STRING()))))));
+ List<Integer> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
read.createDataReader(mock(Split.class), auth)) {
+ reader.forEachRemaining(
+ row -> {
+ assertThat(row.getFieldCount()).isEqualTo(1);
+ result.add(row.getInt(0));
+ });
+ }
+ assertThat(result).containsExactly(1);
+ }
+
+ @Test
+ void testMaskedQueryFilterIsEvaluatedOnceAfterAuthorization() throws
IOException {
+ RowType type = RowType.of(DataTypes.INT(), DataTypes.STRING(),
DataTypes.STRING());
+ TestingDataTableRead read =
+ new TestingDataTableRead(
+ schema(type),
+ GenericRow.of(
+ 0,
+ BinaryString.fromString("wrong"),
+ BinaryString.fromString("match")),
+ GenericRow.of(
+ 1,
+ BinaryString.fromString("wrong"),
+ BinaryString.fromString("match")),
+ GenericRow.of(
+ 2,
+ BinaryString.fromString("MATCH"),
+ BinaryString.fromString("no")));
+ read.withReadType(type.project(new int[] {0}));
+ AtomicInteger evaluations = new AtomicInteger();
+ PredicateBuilder builder = new PredicateBuilder(type);
+ read.withFilter(
+ builder.equal(
+ new CountingFieldTransform(
+ new FieldRef(1, type.getFieldNames().get(1),
DataTypes.STRING()),
+ evaluations),
+ BinaryString.fromString("MATCH")));
+ read.executeFilter();
+ TableQueryAuthResult auth =
+ new TableQueryAuthResult(
+ Collections.singletonList(
+
JsonSerdeUtil.toFlatJson(builder.greaterThan(0, 0))),
+ Collections.singletonMap(
+ type.getFieldNames().get(1),
+ JsonSerdeUtil.toFlatJson(
+ new UpperTransform(
+ Collections.singletonList(
+ new FieldRef(
+ 2,
+
type.getFieldNames().get(2),
+
DataTypes.STRING()))))));
+ List<Integer> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
read.createDataReader(mock(Split.class), auth)) {
+ reader.forEachRemaining(row -> result.add(row.getInt(0)));
+ }
+ assertThat(result).containsExactly(1);
+ // Only the two authorized rows reach the query expression, once each.
+ assertThat(evaluations.get()).isEqualTo(2);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMaskedDisjunctionPreservesUnmaskedOperand(boolean executeFilter)
throws IOException {
+ RowType type =
+ RowType.of(
+ DataTypes.INT(), DataTypes.INT(), DataTypes.STRING(),
DataTypes.STRING());
+ TestingDataTableRead read =
+ new TestingDataTableRead(
+ schema(type),
+ GenericRow.of(
+ 1,
+ 0,
+ BinaryString.fromString("raw"),
+ BinaryString.fromString("match")),
+ GenericRow.of(
+ 11,
+ 0,
+ BinaryString.fromString("raw"),
+ BinaryString.fromString("match")),
+ GenericRow.of(
+ 12,
+ 1,
+ BinaryString.fromString("raw"),
+ BinaryString.fromString("no")),
+ GenericRow.of(
+ 13,
+ 0,
+ BinaryString.fromString("MATCH"),
+ BinaryString.fromString("no")));
+ read.withReadType(type.project(new int[] {0}));
+ PredicateBuilder builder = new PredicateBuilder(type);
+ read.withFilter(
+ PredicateBuilder.and(
+ builder.greaterThan(0, 10),
+ PredicateBuilder.or(
+ builder.equal(2,
BinaryString.fromString("MATCH")),
+ builder.equal(1, 1))));
+ if (executeFilter) {
+ read.executeFilter();
+ }
+ TableQueryAuthResult auth =
+ new TableQueryAuthResult(
+ null,
+ Collections.singletonMap(
+ type.getFieldNames().get(2),
+ JsonSerdeUtil.toFlatJson(
+ new UpperTransform(
+ Collections.singletonList(
+ new FieldRef(
+ 3,
+
type.getFieldNames().get(3),
+
DataTypes.STRING()))))));
+ List<Integer> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
read.createDataReader(mock(Split.class), auth)) {
+ reader.forEachRemaining(
+ row -> {
+ assertThat(row.getFieldCount()).isEqualTo(1);
+ result.add(row.getInt(0));
+ });
+ }
+ // The unmasked standalone conjunct is left to the engine unless full
execution is enabled.
+ assertThat(result)
+ .containsExactlyElementsOf(
+ executeFilter ? Arrays.asList(11, 12) :
Arrays.asList(1, 11, 12));
+ }
+
+ @Test
+ void testReadersKeepTheirOwnAuthorization() throws IOException {
+ RowType type = RowType.of(DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING());
+ TestingDataTableRead read =
+ new TestingDataTableRead(
+ schema(type),
+ GenericRow.of(
+ BinaryString.fromString("raw"),
+ BinaryString.fromString("alpha"),
+ BinaryString.fromString("beta")));
+ read.withReadType(type.project(new int[] {0}));
+ read.withFilter(new PredicateBuilder(type).equal(0,
BinaryString.fromString("ALPHA")))
+ .executeFilter();
+ TableQueryAuthResult firstAuth =
+ new TableQueryAuthResult(
+ null,
+ Collections.singletonMap(
+ type.getFieldNames().get(0),
+ JsonSerdeUtil.toFlatJson(
+ new UpperTransform(
+ Collections.singletonList(
+ new FieldRef(
+ 1,
+
type.getFieldNames().get(1),
+
DataTypes.STRING()))))));
+ TableQueryAuthResult secondAuth =
+ new TableQueryAuthResult(
+ null,
+ Collections.singletonMap(
+ type.getFieldNames().get(0),
+ JsonSerdeUtil.toFlatJson(
+ new UpperTransform(
+ Collections.singletonList(
+ new FieldRef(
+ 2,
+
type.getFieldNames().get(2),
+
DataTypes.STRING()))))));
+ try (RecordReader<InternalRow> first =
read.createDataReader(mock(Split.class), firstAuth);
+ RecordReader<InternalRow> second =
+ read.createDataReader(mock(Split.class), secondAuth)) {
+ List<String> firstRows = new ArrayList<>();
+ first.forEachRemaining(row ->
firstRows.add(row.getString(0).toString()));
+ List<String> secondRows = new ArrayList<>();
+ second.forEachRemaining(row ->
secondRows.add(row.getString(0).toString()));
+ assertThat(firstRows).containsExactly("ALPHA");
+ assertThat(secondRows).isEmpty();
+ }
+ read.withFilter((Predicate) null);
+ List<String> rawRows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
read.createReader(mock(Split.class))) {
+ reader.forEachRemaining(
+ row -> {
+ assertThat(row.getFieldCount()).isEqualTo(1);
+ rawRows.add(row.getString(0).toString());
+ });
+ }
+ assertThat(rawRows).containsExactly("raw");
+ }
+
+ private static class CountingFieldTransform extends FieldTransform {
+
+ private static final long serialVersionUID = 1L;
+ private final AtomicInteger evaluations;
+
+ private CountingFieldTransform(FieldRef field, AtomicInteger
evaluations) {
+ super(field);
+ this.evaluations = evaluations;
+ }
+
+ @Override
+ public Object transform(InternalRow row) {
+ evaluations.incrementAndGet();
+ return super.transform(row);
+ }
+
+ @Override
+ public Transform copyWithNewInputs(List<Object> inputs) {
+ return new CountingFieldTransform((FieldRef) inputs.get(0),
evaluations);
+ }
+ }
+
+ private static TableSchema schema(RowType type) {
+ return new TableSchema(
+ 0,
+ type.getFields(),
+ type.getFieldCount() - 1,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ null);
+ }
+
+ @Test
+ void testExecuteFilterPreservesNestedOutputProjection() throws IOException
{
+ RowType nested =
+ new RowType(
+ Arrays.asList(
+ new DataField(1, "a", DataTypes.INT()),
+ new DataField(2, "b", DataTypes.STRING())));
+ RowType type = new RowType(Collections.singletonList(new DataField(0,
"profile", nested)));
+ RowType outputType =
+ new RowType(
+ Collections.singletonList(
+
type.getFields().get(0).newType(nested.project("b"))));
+ TestingDataTableRead read =
+ new TestingDataTableRead(
+ schema(type),
+ GenericRow.of(GenericRow.of(10,
BinaryString.fromString("kept"))),
+ GenericRow.of((Object) null));
+ read.withReadType(outputType)
+ .withFilter(new PredicateBuilder(type).isNotNull(0))
+ .executeFilter();
+ List<String> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
read.createReader(mock(Split.class))) {
+ reader.forEachRemaining(
+ row -> {
+ InternalRow profile = row.getRow(0, 1);
+ assertThat(profile.getFieldCount()).isEqualTo(1);
+ result.add(profile.getString(0).toString());
+ });
+ }
+ assertThat(result).containsExactly("kept");
+ }
+
+ private static List<String> readRows(TestingDataTableRead read, RowType
outputType)
+ throws IOException {
+ List<String> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
read.createReader(mock(Split.class))) {
+ reader.forEachRemaining(
+ row -> {
+
assertThat(row.getFieldCount()).isEqualTo(outputType.getFieldCount());
+ result.add(row.getString(0) + ":" + row.getInt(1));
+ });
+ }
+ return result;
+ }
+
@Test
void testNoProjectionResetWithoutExplicitReadType() throws IOException {
TableSchema schema =
@@ -119,9 +463,11 @@ class AbstractDataTableReadTest {
private static class TestingDataTableRead extends AbstractDataTableRead {
private RowType appliedReadType;
+ private final List<InternalRow> rows;
- private TestingDataTableRead(TableSchema schema) {
+ private TestingDataTableRead(TableSchema schema, InternalRow... rows) {
super(schema);
+ this.rows = Arrays.asList(rows);
}
@Override
@@ -131,15 +477,16 @@ class AbstractDataTableReadTest {
@Override
public RecordReader<InternalRow> reader(Split split) {
- return new RecordReader<InternalRow>() {
- @Override
- public RecordIterator<InternalRow> readBatch() {
- return null;
- }
-
- @Override
- public void close() {}
- };
+ RowType type = appliedReadType == null ? schema().logicalRowType()
: appliedReadType;
+ NestedProjectedRow projection =
+ NestedProjectedRow.create(schema().logicalRowType(), type);
+ InternalRowSerializer serializer = new InternalRowSerializer(type);
+ List<InternalRow> projected = new ArrayList<>();
+ for (InternalRow row : rows) {
+ projected.add(
+ serializer.copy(projection == null ? row :
projection.replaceRow(row)));
+ }
+ return new IteratorRecordReader<>(projected.iterator());
}
@Override
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala
index 3e34fe56f2..53fcc31c17 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala
@@ -542,7 +542,7 @@ class PermissionProcedureTest extends
PaimonSparkTestWithRestCatalogBase {
|""".stripMargin)
.collect()
}
- assertThat(missingColumn.getMessage).contains("column unknown")
+ assertThat(missingColumn.getMessage).contains("Cannot resolve field
'unknown'")
val missingPrincipal = intercept[Exception] {
spark
@@ -664,7 +664,7 @@ class PermissionProcedureTest extends
PaimonSparkTestWithRestCatalogBase {
.sql("ALTER TABLE paimon.sales.renamed_orders RENAME COLUMN region TO
area")
.collect()
}
- assertThat(renameColumn.getMessage).contains("column region")
+ assertThat(renameColumn.getMessage).contains("Cannot resolve field
'region'")
spark.sql("DROP TABLE paimon.sales.renamed_orders")
spark.sql("""CREATE TABLE paimon.sales.renamed_orders (
