This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 72e4a32b56e00f411dac32452ee5dd485123ecf0 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 (
