diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateVisitor.java index 9f4047b8cb37..b8ca94f57e7f 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateVisitor.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateVisitor.java @@ -40,6 +40,16 @@ static Set collectFieldNames(@Nullable Predicate predicate) { return predicate.visit(new FieldNameCollector()); } + static Set collectTransformFieldNames(Transform transform) { + Set fieldNames = new HashSet<>(); + for (Object input : transform.inputs()) { + if (input instanceof FieldRef) { + fieldNames.add(((FieldRef) input).name()); + } + } + return fieldNames; + } + static Set collectFieldIds(RowType rowType, @Nullable Predicate predicate) { if (predicate == null) { return Collections.emptySet(); @@ -58,13 +68,7 @@ class FieldNameCollector implements PredicateVisitor> { @Override public Set visit(LeafPredicate predicate) { - Set fieldNames = new HashSet<>(); - for (Object input : predicate.transform().inputs()) { - if (input instanceof FieldRef) { - fieldNames.add(((FieldRef) input).name()); - } - } - return fieldNames; + return collectTransformFieldNames(predicate.transform()); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java index 0ee552d887f6..916f63350c7b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java @@ -32,6 +32,7 @@ import org.apache.paimon.rest.exceptions.NotImplementedException; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.SchemaValidation; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.CatalogEnvironment; import org.apache.paimon.table.FileStoreTable; @@ -168,6 +169,8 @@ public static void validateCreateTable(Schema schema, boolean dataTokenEnabled) if (tableType.equals(TableType.FORMAT_TABLE)) { validateFormatTableOptions(options, dataTokenEnabled); } + SchemaValidation.validateQueryAuthTableType( + tableType, options.get(CoreOptions.QUERY_AUTH_ENABLED)); for (DataField field : schema.fields()) { validateDefaultValue(field.type(), field.defaultValue()); } 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 a2a113a5e897..e3fd81442b82 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 @@ -25,26 +25,38 @@ 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.PredicateVisitor; import org.apache.paimon.predicate.Transform; import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.SpecialFields; +import org.apache.paimon.table.Table; import org.apache.paimon.table.source.DataFilePlan; import org.apache.paimon.table.source.QueryAuthSplit; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.TableScan; +import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; import org.apache.paimon.types.RowType; 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 javax.annotation.Nullable; import java.io.Serializable; +import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Collections; +import java.util.Deque; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.Set; import java.util.TreeMap; import java.util.stream.Collectors; @@ -58,6 +70,11 @@ public class TableQueryAuthResult implements Serializable { private final @Nullable List filter; private final @Nullable Map columnMasking; + // Lazily parsed views of the JSON rules; transient so serialization stays unchanged. No + // invalidation needed: an instance is immutable and rebuilt for every plan(). + private transient volatile Optional parsedFilter; + private transient volatile Map parsedMasking; + public TableQueryAuthResult( @Nullable List filter, @Nullable Map columnMasking) { this.filter = filter; @@ -74,8 +91,115 @@ public Map columnMasking() { return columnMasking; } + public boolean hasRules() { + return extractPredicate() != null || !extractColumnMasking().isEmpty(); + } + + /** + * Rejects a search on a query-auth table. Called from the methods that produce a scan or a + * read, not from the builder constructors: the builders are serializable, and deserialization + * would skip a constructor check. + */ + public static void rejectSearchUnderQueryAuth(@Nullable Table table) { + if (table instanceof FileStoreTable + && ((FileStoreTable) table).coreOptions().queryAuthEnabled()) { + throw new UnsupportedOperationException( + "Search is not supported on a query-auth table: the index ranks raw values, " + + "which a column mask invalidates."); + } + } + + /** + * Drops the conjuncts of {@code predicate} referencing any of {@code fields}; returns null when + * nothing remains. Used to keep raw-statistics pushdown off masked columns. + */ + @Nullable + public static Predicate excludeFields(Predicate predicate, Set fields) { + return filterConjuncts(predicate, fields, true); + } + + /** + * Keeps only the conjuncts of {@code predicate} referencing any of {@code fields}; returns null + * when none does. + */ + @Nullable + public static Predicate retainFields(Predicate predicate, Set fields) { + return filterConjuncts(predicate, fields, false); + } + + @Nullable + private static Predicate filterConjuncts( + Predicate predicate, Set fields, boolean keepDisjoint) { + List kept = new ArrayList<>(); + for (Predicate conjunct : PredicateBuilder.splitAnd(predicate)) { + if (Collections.disjoint(PredicateVisitor.collectFieldNames(conjunct), fields) + == keepDisjoint) { + kept.add(conjunct); + } + } + if (kept.isEmpty()) { + return null; + } + // and() folds an always-true/false leaf into a field-less constant, which would drop the + // field names the callers match against; a lone conjunct must come back untouched + return kept.size() == 1 ? kept.get(0) : PredicateBuilder.and(kept); + } + + /** + * Every column read by the conjuncts of {@code filter} that touch {@code maskTargets}. Their + * unmasked operands count too, since splitAnd does not split a disjunction. + */ + private static Set postMaskFilterFields( + @Nullable Predicate filter, Set maskTargets) { + if (filter == null || maskTargets.isEmpty()) { + return Collections.emptySet(); + } + Set maskedInFilter = new HashSet<>(PredicateVisitor.collectFieldNames(filter)); + maskedInFilter.retainAll(maskTargets); + if (maskedInFilter.isEmpty()) { + return Collections.emptySet(); + } + Predicate retained = retainFields(filter, maskedInFilter); + return retained == null + ? Collections.emptySet() + : new HashSet<>(PredicateVisitor.collectFieldNames(retained)); + } + + /** + * The columns a read projecting {@code readFields} under {@code filter} must additionally + * expose: the rule fields, plus the operands of the conjuncts evaluated post-mask. The scan and + * the read both widen by this, and must agree — the read schema is fixed on first use. + */ + public Set authFields(List readFields, @Nullable Predicate filter) { + Set postMask = postMaskFilterFields(filter, extractColumnMasking().keySet()); + // requiredAuthFields de-duplicates, so a plain concatenation is enough here + List visible = + postMask.isEmpty() + ? readFields + : ListUtils.union(readFields, new ArrayList<>(postMask)); + Set ruleFields = requiredAuthFields(visible); + // requiredAuthFields returns what the rules read, not the operands themselves + ruleFields.addAll(postMask); + return ruleFields; + } + + @Nullable + public static RowType appendMissingFields( + RowType tableType, RowType readType, Set ruleFields) { + List 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); + } + public TableScan.Plan convertPlan(TableScan.Plan plan) { - if (filter == null && (columnMasking == null || columnMasking.isEmpty())) { + if (!hasRules()) { return plan; } List authSplits = @@ -87,6 +211,16 @@ public TableScan.Plan convertPlan(TableScan.Plan plan) { @Nullable public Predicate extractPredicate() { + Optional parsed = parsedFilter; + if (parsed == null) { + parsed = Optional.ofNullable(parsePredicate()); + parsedFilter = parsed; + } + return parsed.orElse(null); + } + + @Nullable + private Predicate parsePredicate() { Predicate rowFilter = null; if (filter != null && !filter.isEmpty()) { List predicates = new ArrayList<>(); @@ -116,6 +250,15 @@ public static Predicate remapPredicate(Predicate predicate, RowType rowType) { } public Map extractColumnMasking() { + Map parsed = parsedMasking; + if (parsed == null) { + parsed = parseColumnMasking(); + parsedMasking = parsed; + } + return parsed; + } + + private Map parseColumnMasking() { Map result = new TreeMap<>(); if (columnMasking != null && !columnMasking.isEmpty()) { for (Map.Entry e : columnMasking.entrySet()) { @@ -128,9 +271,154 @@ public Map extractColumnMasking() { result.put(column, transform); } } - return result; + // the cache is shared by every caller, so hand out a view that cannot rewrite the rules + return Collections.unmodifiableMap(result); + } + + /** + * Validates that every column the auth rules reference exists in the table's latest + * schema. A rule keyed by a since-renamed column would silently stop masking; fail closed. + */ + public void validateAgainstSchema(RowType tableType, @Nullable List projectedFields) { + Map masking = extractColumnMasking(); + for (Map.Entry entry : masking.entrySet()) { + String target = entry.getKey(); + // a mask on an unprojected system column never reaches the output; inert, don't reject + if (SpecialFields.SYSTEM_FIELD_NAMES.contains(target) + && (projectedFields == null || !projectedFields.contains(target))) { + continue; + } + checkFieldExists("Column masking", target, tableType, projectedFields); + for (String input : PredicateVisitor.collectTransformFieldNames(entry.getValue())) { + checkFieldExists("Column masking", input, tableType, projectedFields); + // A transform reads the raw row, so an input that is itself masked would be + // consumed unmasked and its raw value published through this target. Masking + // the target of another mask is only self-consistent if composed, which the + // read does not do; refuse the pair rather than leak. + checkArgument( + input.equals(target) || !masking.containsKey(input), + "Column masking on '%s' reads column '%s', which is masked " + + "too. The mask would be computed from the raw value " + + "of '%s' and expose it through '%s'.", + target, + input, + input, + target); + } + } + for (String operand : PredicateVisitor.collectFieldNames(extractPredicate())) { + checkFieldExists("Row filter", operand, tableType, projectedFields); + } + } + + /** + * Fails closed when the read schema exposes a rule's column under a different name, where + * enforcing by name would skip it. A column absent from that schema stays inert. + */ + public void validateReadableWithoutRename(RowType latestType, RowType readType) { + for (Map.Entry entry : extractColumnMasking().entrySet()) { + checkNotRenamed("Column masking", entry.getKey(), latestType, readType); + // a mask whose target is absent from the read schema is inert; skip its inputs + if (readType.containsField(entry.getKey())) { + for (String input : PredicateVisitor.collectTransformFieldNames(entry.getValue())) { + checkNotRenamed("Column masking", input, latestType, readType); + } + } + } + // the row filter is remapped by name as well, so it needs the same binding check + for (String operand : PredicateVisitor.collectFieldNames(extractPredicate())) { + checkNotRenamed("Row filter", operand, latestType, readType); + } } + private static void checkNotRenamed( + String rule, String field, RowType latestType, RowType readType) { + // absent from latest: already thrown by validateAgainstSchema + if (!latestType.containsField(field)) { + return; + } + int id = latestType.getField(field).id(); + if (readType.containsField(field)) { + // a dropped and re-added column keeps the name but gets a fresh id, so the same + // name may be an unrelated column in the snapshot being read + checkArgument( + readType.getField(field).id() == id, + "%s references column '%s' which the snapshot being read exposes " + + "as a different column of the same name (dropped and " + + "re-added since); refusing to read to avoid applying the " + + "rule to unrelated data.", + rule, + field); + return; + } + // not checkArgument: its arguments are evaluated eagerly, and getField(id) throws when + // the id is absent, which is the normal case here + if (readType.containsField(id)) { + throw new IllegalArgumentException( + String.format( + "%s references column '%s' which the snapshot being read exposes as " + + "'%s' (renamed since); refusing to read to avoid applying " + + "the rule by a stale name.", + rule, field, readType.getField(id).name())); + } + } + + /** + * The field names the auth rules read for a query projecting {@code projectedFields}: the + * row-filter operands, plus transitively the inputs of every mask whose target is readable. + */ + private Set requiredAuthFields(List projectedFields) { + Map masking = extractColumnMasking(); + Set ruleFields = new HashSet<>(); + Set readable = new HashSet<>(projectedFields); + Deque newlyReadable = new ArrayDeque<>(readable); + for (String operand : PredicateVisitor.collectFieldNames(extractPredicate())) { + ruleFields.add(operand); + if (readable.add(operand)) { + newlyReadable.add(operand); + } + } + while (!newlyReadable.isEmpty()) { + Transform mask = masking.get(newlyReadable.poll()); + if (mask == null) { + continue; + } + for (String input : PredicateVisitor.collectTransformFieldNames(mask)) { + ruleFields.add(input); + if (readable.add(input)) { + newlyReadable.add(input); + } + } + } + return ruleFields; + } + + private static void checkFieldExists( + String rule, String field, RowType tableType, @Nullable List projectedFields) { + // system fields (e.g. _ROW_ID) cannot be widened in; only usable when projected + if (SpecialFields.SYSTEM_FIELD_NAMES.contains(field)) { + if (projectedFields != null && projectedFields.contains(field)) { + return; + } + throw new IllegalArgumentException( + String.format( + "%s references system column '%s' which the query does not project.", + rule, field)); + } + checkArgument( + tableType.containsField(field), + "%s references column '%s' which does not exist in table schema %s. " + + "The rule may be stale after a column rename or drop; " + + "refusing to read.", + rule, + field, + tableType.getFieldNames()); + } + + /** + * 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. + */ public RecordReader doAuth( RecordReader reader, RowType outputRowType) { return doAuth(reader, outputRowType, extractPredicate(), extractColumnMasking()); @@ -180,10 +468,6 @@ private static InternalRow transform( private static Map transformRemapping( RowType outputRowType, Map masking) { Map out = new HashMap<>(); - if (masking == null || masking.isEmpty()) { - return out; - } - for (Map.Entry e : masking.entrySet()) { String targetColumn = e.getKey(); Transform transform = e.getValue(); diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java index 8929ef4c7bb6..7a16a2e7accb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java @@ -70,6 +70,8 @@ public class DataEvolutionBatchScan implements DataTableScan { private Predicate filter; private TopN topN; private Integer pushDownLimit; + // set when part of the filter reaches the reader only, so limit/TopN must not prune ahead of it + private boolean rowIdFilterDeferred; private RowRangeIndex pushedRowRangeIndex; private GlobalIndexResult globalIndexResult; @@ -90,12 +92,26 @@ public InnerTableScan withFilter(Predicate predicate) { return this; } - Optional> rowRanges = predicate.visit(new RowIdPredicateVisitor()); - if (rowRanges.isPresent()) { - withRowRanges(rowRanges.get()); + // a mask on _ROW_ID makes the predicate's ids the masked ones, so they must not become + // a raw row range; the rules are not known yet, so skip the extraction altogether + if (!queryAuthEnabled()) { + Optional> rowRanges = predicate.visit(new RowIdPredicateVisitor()); + if (rowRanges.isPresent()) { + withRowRanges(rowRanges.get()); + } } this.filter = predicate; + if (queryAuthEnabled()) { + // the wrapped scan defers the filter but strips only masked columns; row ids must + // go here, since data-evolution statistics carry logical columns only + Predicate residual = rowIdSafeResidualFilter(predicate); + rowIdFilterDeferred = containsRowId(predicate); + if (residual != null) { + batchScan.withFilter(residual); + } + return this; + } batchScan.snapshotReader().withFilter(predicate, rowIdSafeResidualFilter(predicate)); return this; } @@ -168,8 +184,8 @@ public InnerTableScan withMetricRegistry(MetricRegistry metricsRegistry) { @Override public InnerTableScan withLimit(int limit) { + // forwarded in plan(), once withFilter has said whether a row-id part was deferred this.pushDownLimit = limit; - batchScan.withLimit(limit); return this; } @@ -279,7 +295,11 @@ public Plan plan() { } } - if (!globalIndexTopNCandidatesFound && topN != null) { + if (pushDownLimit != null && !rowIdFilterDeferred) { + batchScan.withLimit(pushDownLimit); + } + + if (!globalIndexTopNCandidatesFound && topN != null && !rowIdFilterDeferred) { batchScan.withTopN(topN); } @@ -291,7 +311,19 @@ public Plan plan() { return wrapToIndexSplits(splits, rowRangeIndex, scoreGetter); } + private boolean queryAuthEnabled() { + // the table is absent in tests that exercise withFilter in isolation + CoreOptions options = table == null ? null : table.coreOptions(); + return options != null && options.queryAuthEnabled(); + } + private Optional evalGlobalIndex() { + // the index ranks raw values, which a mask may invalidate; fall back to a full scan. + // Checked before the supplied result too: withGlobalIndexResult is public, so a caller + // can hand in one that was computed off the raw values. + if (queryAuthEnabled()) { + return Optional.empty(); + } if (this.globalIndexResult != null) { return Optional.of(globalIndexResult); } 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 88053b03e300..22f70f0cbaed 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 @@ -127,14 +127,14 @@ public DataEvolutionFileStoreScan withFilter(Predicate predicate) { @Override public FileStoreScan withReadType(RowType readType) { - if (readType != null) { - List nonSystemFields = - readType.getFields().stream() - .filter(f -> !SpecialFields.isSystemField(f.id())) - .collect(Collectors.toList()); - if (!nonSystemFields.isEmpty()) { - this.readType = readType; - } + // a type without user columns prunes nothing; assign unconditionally, this method + // may be recalled + if (readType != null + && readType.getFields().stream() + .anyMatch(f -> !SpecialFields.isSystemField(f.id()))) { + this.readType = readType; + } else { + this.readType = null; } return this; } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java index 5e361e45239e..9e66aad52fca 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java @@ -146,10 +146,8 @@ public MergeFileSplitRead withReadType(RowType readType) { readerFactoryBuilder.withReadValueType(adjustedReadType); mergeSorter.setProjectedValueType(adjustedReadType); - // Project away fields added for merging. - if (adjustedReadType != readType) { - outerReadType = readType; - } + // reset rather than latch: this method may be called again + outerReadType = adjustedReadType != readType ? readType : null; return this; } @@ -516,9 +514,11 @@ public RecordReader createNoMergeReader( /** * Returns the pushed read type if {@link #withReadType(RowType)} was called, else the default - * read type. + * read type. This is the value layout the merge, comparator and serializer run on internally; + * when a sequence field was appended for merging, {@link #createMergeReader} projects its + * output back to {@code outerReadType}, so the emitted rows can be narrower than this type. */ - private RowType actualReadType() { + public RowType actualReadType() { return readerFactoryBuilder.readValueType(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index cc0ee88ad702..896612d30599 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -156,6 +156,10 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp validateOnlyContainPrimitiveType(schema.fields(), schema.primaryKeys(), "primary key"); validateOnlyContainPrimitiveType(schema.fields(), schema.partitionKeys(), "partition"); + // reject here rather than only at create time, so ALTER cannot turn the option on for a + // table type that ignores it + validateQueryAuthTableType(options.type(), options.queryAuthEnabled()); + if (options.primaryKeyNullable() && schema.primaryKeys().isEmpty()) { throw new IllegalArgumentException( String.format( @@ -399,6 +403,20 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp validateManifestSort(schema, options); } + /** + * Only a file-store table reads through the auth reader; anywhere else the rules would be + * accepted and then silently not applied. + */ + public static void validateQueryAuthTableType(TableType tableType, boolean queryAuthEnabled) { + checkArgument( + !queryAuthEnabled + || tableType == TableType.TABLE + || tableType == TableType.MATERIALIZED_TABLE, + "%s is not supported on a %s: its read does not apply row filters or column masks.", + CoreOptions.QUERY_AUTH_ENABLED.key(), + tableType); + } + public static void validateFallbackBranch(SchemaManager schemaManager, TableSchema schema) { String fallbackBranch = schema.options().get(CoreOptions.SCAN_FALLBACK_BRANCH.key()); String primaryBranch = schema.options().get(CoreOptions.SCAN_PRIMARY_BRANCH.key()); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index 97058f2b8e03..fa1e00b4ea05 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -290,6 +290,7 @@ public StreamDataTableScan newStreamScan() { DataTableStreamScan scan = new DataTableStreamScan( tableSchema, + schemaManager(), coreOptions(), newSnapshotReader(), snapshotManager(), diff --git a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java index f189d886823f..6f355d6602e5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java @@ -94,6 +94,12 @@ public class LocalTableQuery implements TableQuery { public LocalTableQuery(FileStoreTable table) { this.options = table.coreOptions(); + if (options.queryAuthEnabled()) { + // the lookup cache serves rows straight from the store, past the auth reader + throw new UnsupportedOperationException( + "Local table query is not supported on a query-auth table: it returns raw " + + "values, outside row filters and column masks."); + } this.tableView = new ConcurrentHashMap<>(); FileStore tableStore = table.store(); if (!(tableStore instanceof KeyValueFileStore)) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java index b52424b937b6..1abcfcea306d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java @@ -58,7 +58,6 @@ public abstract class AbstractBatchTableScan extends AbstractDataTableScan { private Integer pushDownLimit; private TopN topN; - private final SchemaManager schemaManager; @Nullable private String readProtectionTagName; protected AbstractBatchTableScan( @@ -67,10 +66,9 @@ protected AbstractBatchTableScan( CoreOptions options, SnapshotReader snapshotReader, TableQueryAuth queryAuth) { - super(schema, options, snapshotReader, queryAuth); + super(schema, schemaManager, options, snapshotReader, queryAuth); this.hasNext = true; - this.schemaManager = schemaManager; if (!schema.primaryKeys().isEmpty() && options.batchScanSkipLevel0()) { // Incremental scans read the delta or changelog files of historical snapshots, which // are always recorded at level 0. Skipping level 0 would drop all of their input. @@ -153,6 +151,10 @@ protected Plan postProcessPlan(Plan plan) { @Override public List listPartitionEntries() { + // partition listing bypasses plan(), so apply the rules here too: without the row filter + // it would report partitions the caller cannot read, and pushing a filter on a masked + // column against raw partition values would drop partitions the query matches + applyAuthRules(); if (startingScanner == null) { startingScanner = createStartingScanner(false); } @@ -224,6 +226,10 @@ private Optional applyPushDownTopN() { } SortValue order = orders.get(0); + if (authMaskedFields.contains(order.field().name())) { + // the pruning below reads raw statistics; a mask may alter the ordering column + return Optional.empty(); + } DataType type = order.field().type(); if (!minmaxAvailable(type)) { return Optional.empty(); 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 9a744a9af0be..7b14dd1cac69 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 @@ -18,12 +18,13 @@ package org.apache.paimon.table.source; +import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.data.InternalRow; import org.apache.paimon.disk.IOManager; -import org.apache.paimon.predicate.FieldRef; 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; @@ -34,7 +35,7 @@ import javax.annotation.Nullable; import java.io.IOException; -import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -42,8 +43,6 @@ import java.util.Optional; import java.util.Set; -import static org.apache.paimon.predicate.PredicateVisitor.collectFieldNames; - /** A {@link InnerTableRead} for data table. */ public abstract class AbstractDataTableRead implements InnerTableRead { @@ -52,8 +51,29 @@ public abstract class AbstractDataTableRead implements InnerTableRead { private Predicate predicate; private final TableSchema schema; - public AbstractDataTableRead(TableSchema schema) { + // reader-level filtering sees raw values, so it stays off for auth-enabled tables, + // 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 resolvedBlobViewFields; + + public AbstractDataTableRead(@Nullable TableSchema schema) { this.schema = schema; + Set blobViewFields = Collections.emptySet(); + boolean queryAuthEnabled = false; + if (schema != null) { + CoreOptions options = CoreOptions.fromMap(schema.options()); + if (options.blobViewResolveEnabled()) { + blobViewFields = options.blobViewField(); + } + queryAuthEnabled = options.queryAuthEnabled(); + } + this.resolvedBlobViewFields = blobViewFields; + this.queryAuthEnabled = queryAuthEnabled; } public abstract void applyReadType(RowType readType); @@ -68,6 +88,9 @@ public TableRead withIOManager(IOManager ioManager) { @Override public final InnerTableRead withFilter(Predicate predicate) { this.predicate = predicate; + if (queryAuthEnabled) { + return this; + } return innerWithFilter(predicate); } @@ -90,6 +113,7 @@ public final InnerTableRead withProjection(int[] projection) { @Override public final InnerTableRead withReadType(RowType readType) { this.readType = readType; + this.appliedReadType = null; applyReadType(readType); return this; } @@ -129,10 +153,11 @@ protected final RecordReader createDataReader( // the underlying reader must retain its own default read type. if (readType != null) { applyReadType(readType); + appliedReadType = null; } RecordReader reader; if (authResult == null) { - reader = reader(split); + reader = backProject(reader(split)); } else { reader = authedReader(split, authResult); } @@ -145,50 +170,130 @@ protected final RecordReader createDataReader( private RecordReader authedReader(Split split, TableQueryAuthResult authResult) throws IOException { - RecordReader reader; - RowType tableType = schema.logicalRowType(); - RowType readType = this.readType == null ? tableType : this.readType; - Predicate authPredicate = authResult.extractPredicate(); - Map columnMasking = authResult.extractColumnMasking(); - ProjectedRow backRow = null; - List readFields = readType.getFieldNames(); - Set readFieldSet = new HashSet<>(readFields); - Map selectedColumnMasking = new HashMap<>(); - for (Map.Entry mask : columnMasking.entrySet()) { - if (readFieldSet.contains(mask.getKey())) { - selectedColumnMasking.put(mask.getKey(), mask.getValue()); - } + List readFields = currentReadType().getFieldNames(); + // masked filter columns are read and masked like rule fields, then evaluated post-mask + Set maskedFilterFields = + maskedFilterFields(authResult.extractColumnMasking().keySet()); + Set ruleFields = authResult.authFields(readFields, predicate); + RowType widened = widenedReadType(authResult, ruleFields); + if (widened != null && !widened.equals(appliedReadType)) { + applyReadType(widened); + appliedReadType = widened; } - Set authFields = new HashSet<>(); - if (authPredicate != null) { - authFields.addAll(collectFieldNames(authPredicate)); - } - for (Map.Entry mask : selectedColumnMasking.entrySet()) { - authFields.add(mask.getKey()); - for (Object input : mask.getValue().inputs()) { - if (input instanceof FieldRef) { - authFields.add(((FieldRef) input).name()); + // 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 masking = authResult.extractColumnMasking(); + Map selectedColumnMasking = Collections.emptyMap(); + if (!masking.isEmpty()) { + Set activeFields = new HashSet<>(readFields); + activeFields.addAll(ruleFields); + selectedColumnMasking = new HashMap<>(); + for (Map.Entry mask : masking.entrySet()) { + if (activeFields.contains(mask.getKey())) { + selectedColumnMasking.put(mask.getKey(), mask.getValue()); } } } - if (!authFields.isEmpty()) { - List expandedFields = new ArrayList<>(readType.getFields()); - for (DataField field : tableType.getFields()) { - if (authFields.contains(field.name()) && !readFieldSet.contains(field.name())) { - expandedFields.add(field); - } + RecordReader reader = + authResult.doAuth( + reader(split), + outputType, + authResult.extractPredicate(), + selectedColumnMasking); + reader = filterMaskedConjuncts(reader, outputType, maskedFilterFields); + return backProject(reader); + } + + private Set maskedFilterFields(Set maskTargets) { + if (predicate == null || maskTargets.isEmpty()) { + return Collections.emptySet(); + } + Set 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 filterMaskedConjuncts( + RecordReader reader, RowType outputType, Set 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 backProject(RecordReader 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 ruleFields) { + RowType tableType = schema.logicalRowType(); + RowType readType = currentReadType(); + Set maskTargets = authResult.extractColumnMasking().keySet(); + for (String name : readType.getFieldNames()) { + if (!ruleFields.contains(name) && !maskTargets.contains(name)) { + continue; + } + if (!tableType.containsField(name)) { + continue; } - if (expandedFields.size() > readType.getFieldCount()) { - readType = readType.copy(expandedFields); - applyReadType(readType); - backRow = ProjectedRow.from(readType.projectIndexes(readFields)); + // 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())); } } - reader = authResult.doAuth(reader(split), readType, authPredicate, selectedColumnMasking); - if (backRow != null) { - reader = reader.transform(backRow::replaceRow); + 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)); + } } - return reader; + return TableQueryAuthResult.appendMissingFields(tableType, readType, ruleFields); } private RecordReader executeFilter(RecordReader reader) { 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 87257a744fae..24bf32e50cf0 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,8 @@ import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateVisitor; +import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.source.snapshot.CompactedStartingScanner; import org.apache.paimon.table.source.snapshot.ContinuousCompactorStartingScanner; @@ -66,10 +68,13 @@ import javax.annotation.Nullable; +import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.Set; import java.util.TimeZone; import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS; @@ -83,6 +88,7 @@ abstract class AbstractDataTableScan implements DataTableScan { private static final Logger LOG = LoggerFactory.getLogger(AbstractDataTableScan.class); protected final TableSchema schema; + protected final SchemaManager schemaManager; private final CoreOptions options; protected final SnapshotReader snapshotReader; private final TableQueryAuth queryAuth; @@ -93,23 +99,53 @@ abstract class AbstractDataTableScan implements DataTableScan { // Whether the auth predicate has a non-partition part (enforced only at read time). Used by // AbstractBatchTableScan to disable limit push down; not pushed through withFilter. protected boolean authHasNonPartitionFilter; + // auth state, refreshed each plan(). The filter is pushed once, without the conjuncts on + // masked columns, whose raw statistics a mask invalidates; the partition fields are + // replaced on each push, as ManifestsReader#withPartitionFilter overwrites rather than ands. + protected Set authMaskedFields = Collections.emptySet(); + @Nullable private RowType appliedScanReadType; + @Nullable private Predicate userFilter; + private boolean filterPushed = false; + private Set pushedMaskedFields = Collections.emptySet(); + private Set partitionFilterFields = Collections.emptySet(); + // re-applied after the deferred filter push, which writes the same slot + @Nullable private Runnable reapplyPartitionFilter; protected AbstractDataTableScan( TableSchema schema, + SchemaManager schemaManager, CoreOptions options, SnapshotReader snapshotReader, TableQueryAuth queryAuth) { this.schema = schema; + this.schemaManager = schemaManager; this.options = options; this.snapshotReader = snapshotReader; this.queryAuth = queryAuth; } - @Override - public final TableScan.Plan plan() { + /** + * Refreshes the auth state and pushes what can be pushed. Every path that reaches the files + * must run this, so it stays the one place the order of these steps is defined. + */ + @Nullable + protected final TableQueryAuthResult applyAuthRules() { TableQueryAuthResult queryAuthResult = authQuery(); // Always apply/clear the auth filter so removing auth leaves no stale partition pruning. applyAuthFilter(queryAuthResult == null ? null : queryAuthResult.extractPredicate()); + this.authMaskedFields = + queryAuthResult == null + ? Collections.emptySet() + : queryAuthResult.extractColumnMasking().keySet(); + rejectMaskedPartitionFilter(); + ensureFilterPushdown(); + return queryAuthResult; + } + + @Override + public final TableScan.Plan plan() { + TableQueryAuthResult queryAuthResult = applyAuthRules(); + applyAuthReadType(queryAuthResult); Plan plan = planWithoutAuth(); if (queryAuthResult != null) { plan = queryAuthResult.convertPlan(plan); @@ -152,7 +188,13 @@ private void applyAuthFilter(@Nullable Predicate authPredicate) { @Override public InnerTableScan withFilter(Predicate predicate) { - snapshotReader.withFilter(predicate); + if (!options.queryAuthEnabled()) { + // no masks to strip; push now, else chain-table sub-scans read before plan() + snapshotReader.withFilter(predicate); + return this; + } + // deferred to plan(), which strips conjuncts on masked columns before stats pruning + this.userFilter = predicate; return this; } @@ -171,38 +213,53 @@ public AbstractDataTableScan withBucketFilter(Filter bucketFilter) { @Override public InnerTableScan withReadType(@Nullable RowType readType) { this.readType = readType; + this.appliedScanReadType = readType; snapshotReader.withReadType(readType); return this; } @Override public AbstractDataTableScan withPartitionFilter(Map partitionSpec) { - snapshotReader.withPartitionFilter(partitionSpec); - return this; + return pushPartitionFilter( + partitionSpec == null + ? Collections.emptySet() + : new HashSet<>(partitionSpec.keySet()), + () -> snapshotReader.withPartitionFilter(partitionSpec)); } @Override public AbstractDataTableScan withPartitionFilter(List partitions) { - snapshotReader.withPartitionFilter(partitions); - return this; + // binary partitions carry no field names; assume every partition key + return pushPartitionFilter( + partitions == null ? Collections.emptySet() : new HashSet<>(schema.partitionKeys()), + () -> snapshotReader.withPartitionFilter(partitions)); } @Override public AbstractDataTableScan withPartitionsFilter(List> partitions) { - snapshotReader.withPartitionsFilter(partitions); - return this; + Set fields = new HashSet<>(); + if (partitions != null) { + partitions.forEach(spec -> fields.addAll(spec.keySet())); + } + return pushPartitionFilter(fields, () -> snapshotReader.withPartitionsFilter(partitions)); } @Override public AbstractDataTableScan withPartitionFilter(PartitionPredicate partitionPredicate) { - snapshotReader.withPartitionFilter(partitionPredicate); - return this; + return pushPartitionFilter( + partitionPredicate == null + ? Collections.emptySet() + : partitionPredicateFields(partitionPredicate), + () -> snapshotReader.withPartitionFilter(partitionPredicate)); } @Override public InnerTableScan withPartitionFilter(Predicate predicate) { - snapshotReader.withPartitionFilter(predicate); - return this; + return pushPartitionFilter( + predicate == null + ? Collections.emptySet() + : PredicateVisitor.collectFieldNames(predicate), + () -> snapshotReader.withPartitionFilter(predicate)); } @Override @@ -221,7 +278,19 @@ protected TableQueryAuthResult authQuery() { if (!options.queryAuthEnabled()) { return null; } - return queryAuth.auth(readType == null ? null : readType.getFieldNames()); + List select = readType == null ? null : readType.getFieldNames(); + TableQueryAuthResult result = queryAuth.auth(select); + if (result != null && result.hasRules()) { + // re-validated every plan, so a schema change under a live scan fails closed + RowType latestSchema = + schemaManager + .latest() + .map(TableSchema::logicalRowType) + .orElseGet(schema::logicalRowType); + result.validateAgainstSchema(latestSchema, select); + result.validateReadableWithoutRename(latestSchema, schema.logicalRowType()); + } + return result; } @Override @@ -242,6 +311,112 @@ public InnerTableScan withRowRangeIndex(RowRangeIndex rowRangeIndex) { return this; } + private AbstractDataTableScan pushPartitionFilter(Set fields, Runnable push) { + partitionFilterFields = fields; + reapplyPartitionFilter = push; + push.run(); + return this; + } + + /** The partition columns a predicate references, or all of them when it cannot be read. */ + private Set partitionPredicateFields(PartitionPredicate partitionPredicate) { + if (partitionPredicate instanceof PartitionPredicate.DefaultPartitionPredicate) { + return PredicateVisitor.collectFieldNames( + ((PartitionPredicate.DefaultPartitionPredicate) partitionPredicate) + .predicate()); + } + return new HashSet<>(schema.partitionKeys()); + } + + /** + * A partition predicate is consumed by pruning and, in Spark, dropped from post-scan + * evaluation, so it can never be re-checked on the masked value. Fail closed. One routed + * through withFilter is fine: that path defers it and evaluates it post-mask. + */ + private void rejectMaskedPartitionFilter() { + if (partitionFilterFields.isEmpty() || authMaskedFields.isEmpty()) { + return; + } + for (String partitionKey : partitionFilterFields) { + if (authMaskedFields.contains(partitionKey)) { + throw new UnsupportedOperationException( + String.format( + "A partition filter cannot be enforced on masked partition key " + + "'%s': engines push partition predicates past the " + + "reader, so it would be matched against the raw " + + "partition value.", + partitionKey)); + } + } + } + + /** + * Pushes the query filter once, minus the conjuncts on masked columns. Also called by partition + * listing; a mask found later on an already-pushed column fails closed. + */ + private void ensureFilterPushdown() { + if (userFilter == null) { + return; + } + Set maskedInFilter = new HashSet<>(authMaskedFields); + maskedInFilter.retainAll(PredicateVisitor.collectFieldNames(userFilter)); + if (!maskedInFilter.isEmpty()) { + // masked conjuncts drop rows at read time only: keep limit/TopN pruning off + authHasNonPartitionFilter = true; + } + if (!filterPushed) { + Predicate effective = + maskedInFilter.isEmpty() + ? userFilter + : TableQueryAuthResult.excludeFields(userFilter, maskedInFilter); + snapshotReader.withFilter(userFilter, effective); + filterPushed = true; + pushedMaskedFields = maskedInFilter; + if (reapplyPartitionFilter != null) { + reapplyPartitionFilter.run(); + } + } else if (!pushedMaskedFields.containsAll(maskedInFilter)) { + throw new IllegalStateException( + "Query auth rules changed and now mask a pushed-down filter column. " + + "Recreate the scan to apply the new rules."); + } + } + + /** + * 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. + */ + 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( + appliedScanReadType, + desired, + new HashSet<>(appliedScanReadType.getFieldNames())); + if (widenedToApplied != null) { + desired = widenedToApplied; + } + if (!desired.equals(appliedScanReadType)) { + snapshotReader.withReadType(desired); + appliedScanReadType = desired; + } + } + public SnapshotReader snapshotReader() { return snapshotReader; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java index 81916cac7bcf..fd02d0011c40 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java @@ -18,6 +18,7 @@ package org.apache.paimon.table.source; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; @@ -125,6 +126,7 @@ public BatchVectorSearchBuilder withOption(String key, String value) { @Override public VectorScan newVectorScan() { + TableQueryAuthResult.rejectSearchUnderQueryAuth(table); if (isPrimaryKeyVectorSearch()) { return new PrimaryKeyVectorScan( table, @@ -138,6 +140,7 @@ public VectorScan newVectorScan() { @Override public BatchVectorRead newBatchVectorRead() { + TableQueryAuthResult.rejectSearchUnderQueryAuth(table); checkArgument(limit > 0, "Limit must be positive, set via withLimit()"); checkNotNull(vectorColumn, "Vector column must be set via withVectorColumn()"); checkArgument( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java index 8b5031de4c1e..02806cfe33fa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java @@ -24,6 +24,7 @@ import org.apache.paimon.consumer.Consumer; import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.BucketMode; import org.apache.paimon.table.source.snapshot.AllDeltaFollowUpScanner; @@ -77,6 +78,7 @@ public class DataTableStreamScan extends AbstractDataTableScan implements Stream public DataTableStreamScan( TableSchema schema, + SchemaManager schemaManager, CoreOptions options, SnapshotReader snapshotReader, SnapshotManager snapshotManager, @@ -84,7 +86,7 @@ public DataTableStreamScan( boolean supportStreamingReadOverwrite, TableQueryAuth queryAuth, boolean hasPk) { - super(schema, options, snapshotReader, queryAuth); + super(schema, schemaManager, options, snapshotReader, queryAuth); this.options = options; this.scanMode = options.toConfiguration().get(CoreOptions.STREAM_SCAN_MODE); @@ -104,7 +106,6 @@ public DataTableStreamScan( @Override public DataTableStreamScan withFilter(Predicate predicate) { super.withFilter(predicate); - snapshotReader.withFilter(predicate); return this; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java index dccd3d7da20f..77d74c9a5f6c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java @@ -19,6 +19,7 @@ package org.apache.paimon.table.source; import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition; import org.apache.paimon.index.pk.PrimaryKeyIndexDefinitions; import org.apache.paimon.partition.PartitionPredicate; @@ -72,6 +73,7 @@ public FullTextSearchBuilder withQuery(String fieldName, String query) { @Override public FullTextScan newFullTextScan() { + TableQueryAuthResult.rejectSearchUnderQueryAuth(table); DataField textColumn = textColumn(); Optional definition = primaryKeyFullTextDefinition(textColumn); return definition.isPresent() @@ -86,6 +88,7 @@ public FullTextScan newFullTextScan() { @Override public FullTextRead newFullTextRead() { + TableQueryAuthResult.rejectSearchUnderQueryAuth(table); checkArgument(limit > 0, "Limit must be positive, set via withLimit()"); DataField textColumn = textColumn(); Optional definition = primaryKeyFullTextDefinition(textColumn); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java index 7d3725444d7d..35460de7b472 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java @@ -19,6 +19,7 @@ package org.apache.paimon.table.source; import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.HybridSearchRanker; @@ -128,6 +129,7 @@ public HybridSearchBuilder withWeightedScoreRanker() { @Override public List routeBuilders() { + TableQueryAuthResult.rejectSearchUnderQueryAuth(table); validateSearch(); Snapshot snapshot = null; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java index e50ff3772d29..fa4037f63306 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java @@ -19,6 +19,7 @@ package org.apache.paimon.table.source; import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileHandler; @@ -116,6 +117,11 @@ protected Plan preProcessPlan() { if (globalIndexSplitResult == null) { return null; } + if (options().queryAuthEnabled()) { + // the splits were selected from raw index values, which a mask may invalidate; + // plan normally instead, as the index evaluation below already does + return null; + } if (globalIndexSplitResult.snapshotId() > 0) { maybeCreateReadProtectionTag(globalIndexSplitResult.snapshotId()); } @@ -135,8 +141,15 @@ protected Plan postProcessPlan(Plan dataPlan) { || snapshotPlan.splits().isEmpty()) { return dataPlan; } + // the index is built over raw values, so a conjunct on a masked column must not reach it + Predicate indexCandidate = + filter == null || authMaskedFields.isEmpty() + ? filter + : TableQueryAuthResult.excludeFields(filter, authMaskedFields); Predicate indexFilter = - filter == null ? null : filter.visit(indexPredicateExtractor).orElse(null); + indexCandidate == null + ? null + : indexCandidate.visit(indexPredicateExtractor).orElse(null); if (indexFilter == null) { return dataPlan; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java index 9d2be281b2c7..ee6949eddb0c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java @@ -245,10 +245,10 @@ public TableRead newRead() { read.withReadType(readType); } if (queryAuthEnabled) { - // Skip TopN (engine re-applies it); apply the limit after auth only without a TopN, - // else an unordered limit could drop sorted rows. + // the reader does not evaluate the query filter on an auth-enabled table, so + // capping the rows here would cut away rows that actually match if (topN == null && limit != null) { - return new LimitTableRead(read, limit); + return new LimitTableRead(read, limit, filter != null); } return read; } @@ -291,10 +291,15 @@ private static class LimitTableRead implements TableRead { private final TableRead delegate; private final int limit; + // with a filter the reader only evaluates it once executeFilter() is requested; + // otherwise the engine does, after this limit, so capping here would drop matches + private final boolean filterPresent; + private boolean filterExecutedByReader = false; - private LimitTableRead(TableRead delegate, int limit) { + private LimitTableRead(TableRead delegate, int limit, boolean filterPresent) { this.delegate = delegate; this.limit = limit; + this.filterPresent = filterPresent; } @Override @@ -306,6 +311,7 @@ public TableRead withMetricRegistry(MetricRegistry registry) { @Override public TableRead executeFilter() { delegate.executeFilter(); + this.filterExecutedByReader = true; return this; } @@ -338,6 +344,9 @@ public RecordReader createReader(TableScan.Plan plan) throws IOExce } private RecordReader limit(RecordReader reader) { + if (filterPresent && !filterExecutedByReader) { + return reader; + } // Stop reading once the limit is reached (return EOF), rather than filtering and // draining the rest of the data. return new RecordReader() { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java index d5c2526a72c0..b5e572b8752b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java @@ -19,6 +19,7 @@ package org.apache.paimon.table.source; import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; @@ -128,6 +129,7 @@ public VectorSearchBuilder withOption(String key, String value) { @Override public VectorScan newVectorScan() { + rejectUnderQueryAuth(); if (isPrimaryKeyVectorSearch()) { return new PrimaryKeyVectorScan( table, @@ -143,6 +145,7 @@ public VectorScan newVectorScan() { @Override public VectorRead newVectorRead() { + rejectUnderQueryAuth(); checkNotNull(vector, "vector must be set via withVector()"); if (isPrimaryKeyVectorSearch()) { return new PrimaryKeyVectorRead(table, vectorColumn, vector, limit, options, filter); @@ -160,4 +163,8 @@ public VectorSearchBuilderImpl withSnapshot(Snapshot snapshot) { this.pinnedSnapshot = snapshot; return this; } + + protected void rejectUnderQueryAuth() { + TableQueryAuthResult.rejectSearchUnderQueryAuth(table); + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java index d8c50d596bab..85e8bd267e1a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java @@ -113,8 +113,8 @@ public RecordReader createReader(Split s) throws IOException { mergeRead.mergeSorter(), forceKeepDelete); if (readType != null) { - ProjectedRow projectedRow = - ProjectedRow.from(readType, mergeRead.tableSchema().logicalRowType()); + // project from the merge read's actual output, which may itself be projected + ProjectedRow projectedRow = ProjectedRow.from(readType, mergeRead.actualReadType()); reader = reader.transform(kv -> kv.replaceValue(projectedRow.replaceRow(kv.value()))); } return KeyValueTableRead.unwrap(reader, mergeRead.tableSchema().options()); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java index 11bb7a353c8a..e6b7f6de7f93 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java @@ -146,6 +146,7 @@ public StreamDataTableScan newStreamScan() { } return new DataTableStreamScan( wrapped.schema(), + schemaManager(), coreOptions(), newSnapshotReader(), snapshotManager(), diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java index 3bc083f3e24f..776730946850 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java @@ -94,9 +94,25 @@ public class SystemTableLoader { public static final List GLOBAL_SYSTEM_TABLES = Arrays.asList(ALL_TABLES, ALL_PARTITIONS, ALL_TABLE_OPTIONS, CATALOG_OPTIONS); + /** + * System tables built from raw metadata -- file names, row counts, per-column min/max and + * distinct/null counts -- none of which a column mask covers. + */ + private static final List PHYSICAL_METADATA_TABLES = + Arrays.asList(FILES, FILE_KEY_RANGES, BINLOG, STATISTICS); + @Nullable public static Table load(String type, FileStoreTable dataTable) { - return Optional.ofNullable(SYSTEM_TABLE_LOADERS.get(type.toLowerCase())) + String name = type.toLowerCase(); + if (PHYSICAL_METADATA_TABLES.contains(name) && dataTable.coreOptions().queryAuthEnabled()) { + throw new UnsupportedOperationException( + String.format( + "System table '%s' is not supported on a query-auth table: it reports " + + "raw file statistics, which column masking cannot be applied " + + "to.", + name)); + } + return Optional.ofNullable(SYSTEM_TABLE_LOADERS.get(name)) .map(f -> f.apply(dataTable)) .orElse(null); } diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java index 3d45924871f1..5b7f64fe8a59 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java @@ -18,14 +18,25 @@ package org.apache.paimon.catalog; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.predicate.ConcatWsTransform; +import org.apache.paimon.predicate.Equal; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.FieldTransform; +import org.apache.paimon.predicate.LeafPredicate; import org.apache.paimon.predicate.Predicate; +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 java.util.Arrays; import java.util.Collections; +import java.util.Map; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests that malformed query-authorization definitions cannot be silently ignored. */ @@ -84,4 +95,71 @@ void testInvalidColumnMaskFailsClosed() { .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("JSON null"); } + + private static final RowType TABLE_TYPE = + RowType.of( + new org.apache.paimon.types.DataField(0, "display", DataTypes.STRING()), + new org.apache.paimon.types.DataField(1, "extra", DataTypes.STRING())); + + private static String filterJson() { + return JsonSerdeUtil.toFlatJson( + LeafPredicate.of( + new FieldTransform(new FieldRef(1, "extra", DataTypes.STRING())), + Equal.INSTANCE, + Collections.singletonList(BinaryString.fromString("x")))); + } + + private static String maskJson() { + return JsonSerdeUtil.toFlatJson( + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(1, "extra", DataTypes.STRING())))); + } + + @Test + public void testValidateRejectsReAddedColumnOfSameName() { + Map masking = Collections.singletonMap("display", maskJson()); + TableQueryAuthResult result = new TableQueryAuthResult(null, masking); + + assertThatCode(() -> result.validateReadableWithoutRename(TABLE_TYPE, TABLE_TYPE)) + .doesNotThrowAnyException(); + + RowType latest = + RowType.of( + new org.apache.paimon.types.DataField(0, "display", DataTypes.STRING()), + new org.apache.paimon.types.DataField(7, "extra", DataTypes.STRING())); + assertThatThrownBy(() -> result.validateReadableWithoutRename(latest, TABLE_TYPE)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("dropped and re-added"); + } + + @Test + public void testValidateRejectsReAddedColumnForRowFilter() { + TableQueryAuthResult result = + new TableQueryAuthResult(Collections.singletonList(filterJson()), null); + assertThatCode(() -> result.validateReadableWithoutRename(TABLE_TYPE, TABLE_TYPE)) + .doesNotThrowAnyException(); + + RowType latest = + RowType.of( + new org.apache.paimon.types.DataField(0, "display", DataTypes.STRING()), + new org.apache.paimon.types.DataField(7, "extra", DataTypes.STRING())); + assertThatThrownBy(() -> result.validateReadableWithoutRename(latest, TABLE_TYPE)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Row filter") + .hasMessageContaining("dropped and re-added"); + } + + @Test + public void testHasRules() { + assertThat(new TableQueryAuthResult(null, null).hasRules()).isFalse(); + assertThat( + new TableQueryAuthResult(Collections.emptyList(), Collections.emptyMap()) + .hasRules()) + .isFalse(); + // a blank entry is now rejected rather than ignored, see testInvalidRowFilterFailsClosed + Map masking = Collections.singletonMap("display", maskJson()); + assertThat(new TableQueryAuthResult(null, masking).hasRules()).isTrue(); + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java index d573b5bd905a..8e6a8541e0a5 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java @@ -18,6 +18,7 @@ package org.apache.paimon.operation; +import org.apache.paimon.CoreOptions; import org.apache.paimon.KeyValue; import org.apache.paimon.TestFileStore; import org.apache.paimon.TestKeyValueGenerator; @@ -44,6 +45,8 @@ import org.apache.paimon.table.PrimaryKeyTableUtils; import org.apache.paimon.table.SpecialFields; import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.IncrementalSplit; +import org.apache.paimon.table.source.splitread.IncrementalDiffSplitRead; import org.apache.paimon.types.BigIntType; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; @@ -323,6 +326,59 @@ record -> record.value().getString(1).toString())); } } + @Test + public void testRepeatedReadTypeResetsOuterProjection() throws Exception { + TestKeyValueGenerator gen = new TestKeyValueGenerator(); + List data = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + data.add(gen.next()); + } + TestFileStore store = + createStore( + TestKeyValueGenerator.DEFAULT_PART_TYPE, + TestKeyValueGenerator.KEY_TYPE, + TestKeyValueGenerator.DEFAULT_ROW_TYPE, + TestKeyValueGenerator.TestKeyValueFieldsExtractor.EXTRACTOR, + DeduplicateMergeFunction.factory(), + Collections.singletonMap(CoreOptions.SEQUENCE_FIELD.key(), "orderId")); + store.commitData(data, gen::getPartition, kv -> 0); + + FileStoreScan scan = store.newScan(); + Long snapshotId = store.snapshotManager().latestSnapshotId(); + Map> filesGroupedByPartition = + scan.withSnapshot(snapshotId).plan().files().stream() + .collect(Collectors.groupingBy(ManifestEntry::partition)); + + int rowsRead = 0; + + MergeFileSplitRead read = store.newRead(); + read.withReadType(TestKeyValueGenerator.DEFAULT_ROW_TYPE.project("shopId", "dt", "hr")); + read.withReadType( + TestKeyValueGenerator.DEFAULT_ROW_TYPE.project("shopId", "dt", "hr", "orderId")); + + for (Map.Entry> entry : filesGroupedByPartition.entrySet()) { + RecordReader reader = + read.createReader( + DataSplit.builder() + .withSnapshot(snapshotId) + .withPartition(entry.getKey()) + .withBucket(0) + .withDataFiles( + entry.getValue().stream() + .map(ManifestEntry::file) + .collect(Collectors.toList())) + .withBucketPath("not used") + .build()); + RecordReaderIterator iterator = new RecordReaderIterator<>(reader); + while (iterator.hasNext()) { + assertThat(iterator.next().value().getFieldCount()).isEqualTo(4); + rowsRead++; + } + iterator.close(); + } + assertThat(rowsRead).isPositive(); + } + @Test public void testPostponeReader() throws Exception { RowType keyType = @@ -391,6 +447,70 @@ private static KeyValue keyValue( return new KeyValue().replace(GenericRow.of(key), sequenceNumber, kind, row); } + @Test + public void testIncrementalDiffReadOnProjectedMergeRead() throws Exception { + TestKeyValueGenerator gen = new TestKeyValueGenerator(); + List before = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + before.add(gen.next()); + } + List after = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + after.add(gen.next()); + } + TestFileStore store = + createStore( + TestKeyValueGenerator.DEFAULT_PART_TYPE, + TestKeyValueGenerator.KEY_TYPE, + TestKeyValueGenerator.DEFAULT_ROW_TYPE, + TestKeyValueGenerator.TestKeyValueFieldsExtractor.EXTRACTOR, + DeduplicateMergeFunction.factory()); + store.commitData(before, gen::getPartition, kv -> 0); + store.commitData(after, gen::getPartition, kv -> 0); + + FileStoreScan scan = store.newScan(); + Long snapshotId = store.snapshotManager().latestSnapshotId(); + Map> filesByPartition = + scan.withSnapshot(snapshotId).plan().files().stream() + .collect(Collectors.groupingBy(ManifestEntry::partition)); + + MergeFileSplitRead mergeRead = store.newRead(); + RowType projection = TestKeyValueGenerator.DEFAULT_ROW_TYPE.project("shopId", "dt", "hr"); + mergeRead.withReadType(projection); + SplitRead diffRead = new IncrementalDiffSplitRead(mergeRead); + diffRead.withReadType(projection); + + int rowsRead = 0; + + for (Map.Entry> entry : filesByPartition.entrySet()) { + List files = + entry.getValue().stream().map(ManifestEntry::file).collect(Collectors.toList()); + IncrementalSplit split = + new IncrementalSplit( + snapshotId, + entry.getKey(), + 0, + 1, + Collections.emptyList(), + null, + files, + null, + false); + RecordReaderIterator iterator = + new RecordReaderIterator<>(diffRead.createReader(split)); + while (iterator.hasNext()) { + InternalRow row = iterator.next(); + assertThat(row.getFieldCount()).isEqualTo(3); + assertThat(row.getString(1).toString()).hasSize(8); + row.getInt(0); + row.getInt(2); + rowsRead++; + } + iterator.close(); + } + assertThat(rowsRead).isPositive(); + } + private List writeThenRead( List data, RowType readKeyType, @@ -446,7 +566,8 @@ private TestFileStore createStore( KeyValueFieldsExtractor extractor, MergeFunctionFactory mfFactory) throws Exception { - return createStore(1, partitionType, keyType, valueType, extractor, mfFactory); + return createStore( + 1, partitionType, keyType, valueType, extractor, mfFactory, Collections.emptyMap()); } private TestFileStore createStore( @@ -457,6 +578,36 @@ private TestFileStore createStore( KeyValueFieldsExtractor extractor, MergeFunctionFactory mfFactory) throws Exception { + return createStore( + numBuckets, + partitionType, + keyType, + valueType, + extractor, + mfFactory, + Collections.emptyMap()); + } + + private TestFileStore createStore( + RowType partitionType, + RowType keyType, + RowType valueType, + KeyValueFieldsExtractor extractor, + MergeFunctionFactory mfFactory, + Map options) + throws Exception { + return createStore(1, partitionType, keyType, valueType, extractor, mfFactory, options); + } + + private TestFileStore createStore( + int numBuckets, + RowType partitionType, + RowType keyType, + RowType valueType, + KeyValueFieldsExtractor extractor, + MergeFunctionFactory mfFactory, + Map options) + throws Exception { Path path = new Path(tempDir.toUri()); SchemaManager schemaManager = new SchemaManager(FileIOFinder.find(path), path); boolean valueCountMode = mfFactory.create() instanceof TestValueCountMergeFunction; @@ -476,7 +627,7 @@ private TestFileStore createStore( "")), partitionType.getFieldNames().stream()) .collect(Collectors.toList()), - Collections.emptyMap(), + options, null); TableSchema tableSchema = schemaManager.createTable(schema); return new TestFileStore.Builder( 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 87116e3e4576..6d68f64b9b42 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 @@ -31,6 +31,8 @@ import org.apache.paimon.consumer.ConsumerInfo; import org.apache.paimon.consumer.ConsumerManager; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobViewStruct; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.serializer.InternalRowSerializer; @@ -40,9 +42,11 @@ import org.apache.paimon.function.Function; import org.apache.paimon.function.FunctionChange; import org.apache.paimon.function.FunctionDefinition; +import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.operation.BaseAppendFileStoreWrite; import org.apache.paimon.operation.FileStoreWrite; import org.apache.paimon.options.Options; @@ -56,6 +60,7 @@ import org.apache.paimon.predicate.FieldTransform; import org.apache.paimon.predicate.GreaterOrEqual; import org.apache.paimon.predicate.GreaterThan; +import org.apache.paimon.predicate.LeafFunction; import org.apache.paimon.predicate.LeafPredicate; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; @@ -78,9 +83,11 @@ import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.Instant; +import org.apache.paimon.table.SpecialFields; import org.apache.paimon.table.Table; import org.apache.paimon.table.TableSnapshot; import org.apache.paimon.table.object.ObjectTable; +import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.sink.BatchTableWrite; import org.apache.paimon.table.sink.BatchWriteBuilder; @@ -92,11 +99,16 @@ import org.apache.paimon.table.source.InnerTableScan; import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.table.source.Split; +import org.apache.paimon.table.source.StreamTableScan; import org.apache.paimon.table.source.TableRead; +import org.apache.paimon.table.source.TableScan; import org.apache.paimon.table.system.SystemTableLoader; import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.InternalRowUtils; +import org.apache.paimon.utils.Range; import org.apache.paimon.utils.SnapshotManager; import org.apache.paimon.utils.SnapshotNotExistException; import org.apache.paimon.utils.StringUtils; @@ -3899,6 +3911,1172 @@ void testColumnMaskingApplyOnRead() throws Exception { .isEqualTo("value"); // col5 NOT masked - original value } + private Table createMaskingAuthTable( + Identifier identifier, List fields, Map extraOptions) + throws Exception { + return createMaskingAuthTable( + identifier, fields, Collections.emptyList(), Collections.emptyList(), extraOptions); + } + + private Table createMaskingAuthTable( + Identifier identifier, + List fields, + List partitionKeys, + List primaryKeys, + Map extraOptions) + throws Exception { + catalog.createDatabase(identifier.getDatabaseName(), true); + Map options = new HashMap<>(extraOptions); + options.put(QUERY_AUTH_ENABLED.key(), "true"); + catalog.createTable( + identifier, new Schema(fields, partitionKeys, primaryKeys, options, ""), true); + return catalog.getTable(identifier); + } + + private static LeafPredicate leaf( + int index, String name, DataType type, LeafFunction function, Object literal) { + return LeafPredicate.of( + new FieldTransform(new FieldRef(index, name, type)), + function, + Collections.singletonList(literal)); + } + + private void setColumnMask(Identifier identifier, String target, Transform transform) { + Map masking = new HashMap<>(); + masking.put(target, transform); + setColumnMasking(identifier, masking); + } + + /** Masks {@code target} with the constant {@code ****}. */ + private void maskConstant(Identifier identifier, String target) { + setColumnMask( + identifier, + target, + new ConcatTransform(Collections.singletonList(BinaryString.fromString("****")))); + } + + private static List stringFields(String... names) { + List fields = new ArrayList<>(); + for (int i = 0; i < names.length; i++) { + fields.add(new DataField(i, names[i], DataTypes.STRING())); + } + return fields; + } + + private static void writeStringRows(Table table, String[]... rows) throws Exception { + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + BatchTableWrite write = writeBuilder.newWrite(); + for (String[] values : rows) { + Object[] converted = new Object[values.length]; + for (int i = 0; i < values.length; i++) { + converted[i] = BinaryString.fromString(values[i]); + } + write.write(GenericRow.of(converted)); + } + BatchTableCommit commit = writeBuilder.newCommit(); + commit.commit(write.prepareCommit()); + write.close(); + commit.close(); + } + + private static void writeStringRow(Table table, String... values) throws Exception { + writeStringRows(table, values); + } + + private void maskDisplayWithFullName(Identifier identifier) { + setColumnMask( + identifier, + "display", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(0, "first", DataTypes.STRING()), + new FieldRef(1, "last", DataTypes.STRING())))); + } + + private static List collectRows(RecordReader reader, RowType rowType) + throws Exception { + List rows = new ArrayList<>(); + reader.forEachRemaining(row -> rows.add(InternalRowUtils.copyInternalRow(row, rowType))); + return rows; + } + + @Test + void testColumnMaskingCrossColumnWithProjection() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_cross_column"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("first", "last", "display", "other"), + Collections.emptyMap()); + writeStringRows( + table, + new String[] {"john", "doe", "ignored", "o1"}, + new String[] {"jane", "roe", "ignored", "o2"}); + maskDisplayWithFullName(identifier); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {2}); + List splits = readBuilder.newScan().plan().splits(); + List rows = + collectRows( + readBuilder.newRead().createReader(splits), + table.rowType().project("display")); + + assertThat(rows).hasSize(2); + for (InternalRow row : rows) { + assertThat(row.getFieldCount()).isEqualTo(1); + } + assertThat( + rows.stream() + .map(row -> row.getString(0).toString()) + .collect(java.util.stream.Collectors.toList())) + .containsExactlyInAnyOrder("john-doe", "jane-roe"); + } + + @Test + void testColumnMaskingProjectionAcrossMultipleSplits() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_multi_split"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("first", "last", "display"), + Collections.singletonMap( + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b")); + writeStringRow(table, "john", "doe", "ignored"); + writeStringRow(table, "jane", "roe", "ignored"); + maskDisplayWithFullName(identifier); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {2}); + List splits = readBuilder.newScan().plan().splits(); + assertThat(splits.size()).isGreaterThan(1); + List rows = + collectRows( + readBuilder.newRead().createReader(splits), + table.rowType().project("display")); + + List values = new ArrayList<>(); + for (InternalRow row : rows) { + assertThat(row.getFieldCount()).isEqualTo(1); + values.add(row.getString(0).toString()); + } + assertThat(values).containsExactlyInAnyOrder("john-doe", "jane-roe"); + } + + @Test + void testColumnMaskingOnRowFilterColumnWithProjection() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_filter_target"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("first", "last", "display", "other"), + Collections.emptyMap()); + writeStringRow(table, "john", "doe", "secret", "o1"); + + LeafPredicate displayFilter = + leaf( + 2, + "display", + DataTypes.STRING(), + Equal.INSTANCE, + BinaryString.fromString("secret")); + setRowFilter(identifier, Collections.singletonList(displayFilter)); + maskDisplayWithFullName(identifier); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {3}); + List splits = readBuilder.newScan().plan().splits(); + List rows = + collectRows( + readBuilder.newRead().createReader(splits), + table.rowType().project("other")); + + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getFieldCount()).isEqualTo(1); + assertThat(rows.get(0).getString(0).toString()).isEqualTo("o1"); + } + + @Test + void testColumnMaskingRevokedOnSameTableRead() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_revoked"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("first", "last", "display"), + Collections.emptyMap()); + writeStringRow(table, "john", "doe", "plain"); + maskDisplayWithFullName(identifier); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {2}); + TableRead read = readBuilder.newRead(); + List masked = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("display")); + assertThat(masked).hasSize(1); + assertThat(masked.get(0).getString(0).toString()).isEqualTo("john-doe"); + + setColumnMasking(identifier, new HashMap<>()); + List plain = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("display")); + assertThat(plain).hasSize(1); + assertThat(plain.get(0).getFieldCount()).isEqualTo(1); + assertThat(plain.get(0).getString(0).toString()).isEqualTo("plain"); + } + + @Test + void testColumnMaskingGrantedAfterReadSchemaFixed() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_granted"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("first", "last", "display"), + Collections.emptyMap()); + writeStringRow(table, "john", "doe", "plain"); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {2}); + TableRead read = readBuilder.newRead(); + List plain = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("display")); + assertThat(plain.get(0).getString(0).toString()).isEqualTo("plain"); + + // a mask granted afterwards needs columns outside that projection: the same reader + // widens for this split and still emits only the projected column + maskDisplayWithFullName(identifier); + List masked = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("display")); + assertThat(masked.get(0).getFieldCount()).isEqualTo(1); + assertThat(masked.get(0).getString(0).toString()).isEqualTo("john-doe"); + } + + @Test + void testColumnMaskingPreservesNestedProjection() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_nested"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "display", DataTypes.STRING())); + fields.add( + new DataField( + 1, + "s", + DataTypes.ROW( + new DataField(2, "a", DataTypes.STRING()), + new DataField(3, "b", DataTypes.STRING())))); + fields.add(new DataField(4, "extra", DataTypes.STRING())); + Table table = createMaskingAuthTable(identifier, fields, Collections.emptyMap()); + + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + BatchTableWrite write = writeBuilder.newWrite(); + write.write( + GenericRow.of( + BinaryString.fromString("ignored"), + GenericRow.of(BinaryString.fromString("AV"), BinaryString.fromString("BV")), + BinaryString.fromString("EX"))); + BatchTableCommit commit = writeBuilder.newCommit(); + commit.commit(write.prepareCommit()); + write.close(); + commit.close(); + + RowType tableRowType = table.rowType(); + DataField sField = tableRowType.getField("s"); + RowType prunedS = ((RowType) sField.type()).project("b"); + RowType prunedReadType = + new RowType( + Arrays.asList( + tableRowType.getField("display"), + new DataField(sField.id(), "s", prunedS))); + + ReadBuilder readBuilder = table.newReadBuilder().withReadType(prunedReadType); + List rows = + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + prunedReadType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getRow(1, 1).getString(0).toString()).isEqualTo("BV"); + + setColumnMask( + identifier, + "display", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(4, "extra", DataTypes.STRING())))); + + readBuilder = table.newReadBuilder().withReadType(prunedReadType); + rows = + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + prunedReadType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getFieldCount()).isEqualTo(2); + assertThat(rows.get(0).getString(0).toString()).isEqualTo("EX"); + assertThat(rows.get(0).getRow(1, 1).getString(0).toString()).isEqualTo("BV"); + } + + @Test + void testColumnMaskingStaleRuleFailsClosed() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_stale"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("first", "last", "display"), + Collections.emptyMap()); + writeStringRow(table, "john", "doe", "secret"); + + maskConstant(identifier, "renamed_away"); + assertThatThrownBy(() -> readFully(table)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not exist in table schema"); + + setColumnMask( + identifier, + "display", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(0, "ghost", DataTypes.STRING())))); + assertThatThrownBy(() -> readFully(table)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not exist in table schema"); + } + + @Test + void testColumnMaskingRuleChangeOnRetainedColumn() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_retained"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("display", "hidden_a", "hidden_b"), + Collections.emptyMap()); + writeStringRow(table, "d1", "a1", "b1"); + + Map rules = new HashMap<>(); + rules.put( + "display", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(1, "hidden_a", DataTypes.STRING())))); + setColumnMasking(identifier, rules); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {0}); + TableRead read = readBuilder.newRead(); + List masked = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("display")); + assertThat(masked.get(0).getString(0).toString()).isEqualTo("a1"); + + rules.clear(); + rules.put( + "hidden_a", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(2, "hidden_b", DataTypes.STRING())))); + setColumnMasking(identifier, rules); + List plain = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("display")); + assertThat(plain).hasSize(1); + assertThat(plain.get(0).getFieldCount()).isEqualTo(1); + assertThat(plain.get(0).getString(0).toString()).isEqualTo("d1"); + } + + @Test + void testColumnMaskingRejectsNestedPrunedMaskTarget() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_pruned_target"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "display", DataTypes.STRING())); + fields.add( + new DataField( + 1, + "s", + DataTypes.ROW( + new DataField(2, "a", DataTypes.STRING()), + new DataField(3, "b", DataTypes.STRING())))); + Table table = createMaskingAuthTable(identifier, fields, Collections.emptyMap()); + + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + BatchTableWrite write = writeBuilder.newWrite(); + write.write( + GenericRow.of( + BinaryString.fromString("d1"), + GenericRow.of( + BinaryString.fromString("AV"), BinaryString.fromString("BV")))); + BatchTableCommit commit = writeBuilder.newCommit(); + commit.commit(write.prepareCommit()); + write.close(); + commit.close(); + + RowType tableRowType = table.rowType(); + DataField sField = tableRowType.getField("s"); + maskConstant(identifier, "s"); + + RowType prunedS = ((RowType) sField.type()).project("b"); + RowType prunedReadType = + new RowType( + Arrays.asList( + tableRowType.getField("display"), + new DataField(sField.id(), "s", prunedS))); + ReadBuilder readBuilder = table.newReadBuilder().withReadType(prunedReadType); + assertThatThrownBy( + () -> + collectRows( + readBuilder + .newRead() + .createReader( + readBuilder.newScan().plan().splits()), + prunedReadType)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("pruned"); + } + + @Test + void testColumnMaskingOnColumnAddedAfterSnapshot() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_time_travel"); + Table table = + createMaskingAuthTable( + identifier, stringFields("first", "display"), Collections.emptyMap()); + writeStringRow(table, "john", "d1"); // snapshot 1 + + catalog.alterTable( + identifier, + Collections.singletonList(SchemaChange.addColumn("extra", DataTypes.STRING())), + false); + maskConstant(identifier, "extra"); + + Table latest = catalog.getTable(identifier); + ReadBuilder latestRead = latest.newReadBuilder(); + List latestRows = + collectRows( + latestRead.newRead().createReader(latestRead.newScan().plan().splits()), + latest.rowType()); + assertThat(latestRows).hasSize(1); + assertThat(latestRows.get(0).getString(2).toString()).isEqualTo("****"); + + Table old = + catalog.getTable(identifier) + .copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1")); + ReadBuilder oldRead = old.newReadBuilder(); + List oldRows = + collectRows( + oldRead.newRead().createReader(oldRead.newScan().plan().splits()), + old.rowType()); + assertThat(oldRows).hasSize(1); + assertThat(oldRows.get(0).getString(0).toString()).isEqualTo("john"); + } + + @Test + void testColumnMaskingRenamedColumnTimeTravelFailsClosed() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_rename_travel"); + Table table = + createMaskingAuthTable( + identifier, stringFields("first", "secret"), Collections.emptyMap()); + writeStringRow(table, "john", "s1"); // snapshot 1, column named "secret" + + catalog.alterTable( + identifier, + Collections.singletonList(SchemaChange.renameColumn("secret", "masked_secret")), + false); + maskConstant(identifier, "masked_secret"); + + Table latest = catalog.getTable(identifier); + ReadBuilder latestRead = latest.newReadBuilder(); + List latestRows = + collectRows( + latestRead.newRead().createReader(latestRead.newScan().plan().splits()), + latest.rowType()); + assertThat(latestRows.get(0).getString(1).toString()).isEqualTo("****"); + + Table old = + catalog.getTable(identifier) + .copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1")); + ReadBuilder oldRead = old.newReadBuilder(); + assertThatThrownBy(() -> oldRead.newScan().plan()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("renamed"); + } + + @Test + void testColumnMaskingSystemTargetInertWhenUnprojected() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_system_target"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("display", "other"), + Collections.singletonMap(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")); + writeStringRow(table, "d1", "o1"); + + maskConstant(identifier, "_ROW_ID"); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {0}); + List rows = + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("display")); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getString(0).toString()).isEqualTo("d1"); + } + + @Test + void testColumnMaskingInertTargetWithRenamedInputTimeTravel() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_inert_input"); + Table table = + createMaskingAuthTable( + identifier, stringFields("first", "old_input"), Collections.emptyMap()); + writeStringRow(table, "john", "in1"); // snapshot 1 + + catalog.alterTable( + identifier, + Collections.singletonList(SchemaChange.renameColumn("old_input", "renamed_input")), + false); + catalog.alterTable( + identifier, + Collections.singletonList(SchemaChange.addColumn("display", DataTypes.STRING())), + false); + setColumnMask( + identifier, + "display", + new FieldTransform(new FieldRef(1, "renamed_input", DataTypes.STRING()))); + + Table old = + catalog.getTable(identifier) + .copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1")); + ReadBuilder oldRead = old.newReadBuilder(); + List rows = + collectRows( + oldRead.newRead().createReader(oldRead.newScan().plan().splits()), + old.rowType()); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getString(0).toString()).isEqualTo("john"); + assertThat(rows.get(0).getString(1).toString()).isEqualTo("in1"); + } + + @Test + void testColumnMaskingRevalidatedAfterRulesDisappear() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_revalidate"); + Table table = + createMaskingAuthTable( + identifier, stringFields("first", "secret"), Collections.emptyMap()); + writeStringRow(table, "john", "s1"); + + StreamTableScan scan = table.newReadBuilder().newStreamScan(); + + Map masking = new HashMap<>(); + masking.put( + "secret", + new ConcatTransform(Collections.singletonList(BinaryString.fromString("****")))); + setColumnMasking(identifier, masking); + scan.plan(); + + setColumnMasking(identifier, Collections.emptyMap()); + scan.plan(); + + catalog.alterTable( + identifier, + Collections.singletonList(SchemaChange.renameColumn("secret", "hidden")), + false); + setColumnMasking(identifier, masking); + assertThatThrownBy(scan::plan) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not exist in table schema"); + } + + @Test + void testColumnMaskingRenamedUnderLiveScanFailsClosed() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_live_rename"); + Table table = + createMaskingAuthTable( + identifier, stringFields("first", "secret"), Collections.emptyMap()); + writeStringRow(table, "john", "s1"); + + maskConstant(identifier, "secret"); + + StreamTableScan scan = table.newReadBuilder().newStreamScan(); + scan.plan(); + + catalog.alterTable( + identifier, + Collections.singletonList(SchemaChange.renameColumn("secret", "hidden")), + false); + assertThatThrownBy(scan::plan) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not exist in table schema"); + } + + @Test + void testColumnMaskingRejectsNestedPrunedRuleInput() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_pruned_input"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "display", DataTypes.STRING())); + fields.add( + new DataField( + 1, + "s", + DataTypes.ROW( + new DataField(2, "a", DataTypes.STRING()), + new DataField(3, "b", DataTypes.STRING())))); + Table table = createMaskingAuthTable(identifier, fields, Collections.emptyMap()); + + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + BatchTableWrite write = writeBuilder.newWrite(); + write.write( + GenericRow.of( + BinaryString.fromString("d1"), + GenericRow.of( + BinaryString.fromString("AV"), BinaryString.fromString("BV")))); + BatchTableCommit commit = writeBuilder.newCommit(); + commit.commit(write.prepareCommit()); + write.close(); + commit.close(); + + RowType tableRowType = table.rowType(); + DataField sField = tableRowType.getField("s"); + setColumnMask( + identifier, + "display", + new CastTransform(new FieldRef(1, "s", sField.type()), DataTypes.STRING())); + + RowType prunedS = ((RowType) sField.type()).project("b"); + RowType prunedReadType = + new RowType( + Arrays.asList( + tableRowType.getField("display"), + new DataField(sField.id(), "s", prunedS))); + ReadBuilder readBuilder = table.newReadBuilder().withReadType(prunedReadType); + assertThatThrownBy( + () -> + collectRows( + readBuilder + .newRead() + .createReader( + readBuilder.newScan().plan().splits()), + prunedReadType)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("pruned"); + } + + @Test + void testColumnMaskingWithDataEvolutionColumnFiles() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_data_evolution"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "f0", DataTypes.INT())); + fields.add(new DataField(1, "f1", DataTypes.STRING())); + fields.add(new DataField(2, "f2", DataTypes.STRING())); + Map options = new HashMap<>(); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + Table table = createMaskingAuthTable(identifier, fields, options); + + RowType tableRowType = table.rowType(); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write0 = + builder.newWrite().withWriteType(tableRowType.project("f0", "f1"))) { + write0.write(GenericRow.of(0, BinaryString.fromString("a0"))); + write0.write(GenericRow.of(1, BinaryString.fromString("a1"))); + builder.newCommit().commit(write0.prepareCommit()); + } + long rowId = ((FileStoreTable) table).snapshotManager().latestSnapshot().nextRowId() - 2; + builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write1 = + builder.newWrite().withWriteType(tableRowType.project("f2"))) { + write1.write(GenericRow.of(BinaryString.fromString("b0"))); + write1.write(GenericRow.of(BinaryString.fromString("b1"))); + List commitables = write1.prepareCommit(); + for (CommitMessage c : commitables) { + CommitMessageImpl message = (CommitMessageImpl) c; + List newFiles = + new ArrayList<>(message.newFilesIncrement().newFiles()); + message.newFilesIncrement().newFiles().clear(); + for (DataFileMeta file : newFiles) { + message.newFilesIncrement().newFiles().add(file.assignFirstRowId(rowId)); + } + } + builder.newCommit().commit(commitables); + } + + setColumnMask( + identifier, + "f1", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(2, "f2", DataTypes.STRING())))); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {1}); + List rows = + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + tableRowType.project("f1")); + assertThat(rows).hasSize(2); + assertThat( + rows.stream() + .map(row -> row.isNullAt(0) ? null : row.getString(0).toString()) + .collect(java.util.stream.Collectors.toList())) + .containsExactlyInAnyOrder("b0", "b1"); + } + + @Test + void testColumnMaskingRejectsUnprojectedBlobViewInput() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_blob_view"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "display", DataTypes.STRING())); + fields.add(new DataField(1, "image", DataTypes.BLOB())); + Map options = new HashMap<>(); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + options.put(CoreOptions.BLOB_FIELD.key(), "image"); + options.put(CoreOptions.BLOB_VIEW_FIELD.key(), "image"); + options.put(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED.key(), "true"); + Table table = createMaskingAuthTable(identifier, fields, options); + + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + BatchTableWrite write = writeBuilder.newWrite(); + write.write( + GenericRow.of( + BinaryString.fromString("d1"), + Blob.fromView(new BlobViewStruct(identifier, 1, 0L)))); + BatchTableCommit commit = writeBuilder.newCommit(); + commit.commit(write.prepareCommit()); + write.close(); + commit.close(); + + setColumnMask( + identifier, + "display", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(1, "image", DataTypes.STRING())))); + + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {0}); + assertThatThrownBy( + () -> + collectRows( + readBuilder + .newRead() + .createReader( + readBuilder.newScan().plan().splits()), + table.rowType().project("display"))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("blob-view"); + } + + @Test + void testColumnMaskingDisablesTopNPushdown() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_topn"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "display", DataTypes.INT())); + fields.add(new DataField(1, "source", DataTypes.INT())); + Table table = + createMaskingAuthTable( + identifier, + fields, + Collections.singletonMap( + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b")); + for (int[] row : new int[][] {{100, 0}, {50, 1000}}) { + commitRows(table, GenericRow.of(row[0], row[1])); + } + + setColumnMask( + identifier, + "display", + new FieldTransform(new FieldRef(1, "source", DataTypes.INT()))); + + TopN topN = + new TopN(new FieldRef(0, "display", DataTypes.INT()), DESCENDING, NULLS_LAST, 1); + ReadBuilder readBuilder = + table.newReadBuilder().withProjection(new int[] {0}).withTopN(topN); + List splits = readBuilder.newScan().plan().splits(); + List rows = + collectRows( + readBuilder.newRead().createReader(splits), + table.rowType().project("display")); + assertThat( + rows.stream() + .map(row -> row.getInt(0)) + .collect(java.util.stream.Collectors.toList())) + .contains(1000); + } + + @Test + void testColumnMaskingReadingSystemField() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_row_id"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "display", DataTypes.BIGINT())); + Table table = + createMaskingAuthTable( + identifier, + fields, + Collections.singletonMap(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")); + + commitRows(table, GenericRow.of(42L)); + + setColumnMask( + identifier, + "display", + new FieldTransform(new FieldRef(0, "_ROW_ID", DataTypes.BIGINT().notNull()))); + + RowType readType = + new RowType( + Arrays.asList( + table.rowType().getField("display"), + org.apache.paimon.table.SpecialFields.ROW_ID)); + ReadBuilder readBuilder = table.newReadBuilder().withReadType(readType); + List rows = + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + readType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(0)).isEqualTo(rows.get(0).getLong(1)); + } + + @Test + void testColumnMaskingSystemFieldValidation() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_system_validation"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("display", "other"), + Collections.singletonMap(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")); + writeStringRow(table, "d1", "o1"); + + Map masking = new HashMap<>(); + masking.put( + "_KEY_display", + new ConcatTransform(Collections.singletonList(BinaryString.fromString("****")))); + setColumnMasking(identifier, masking); + assertThatThrownBy(() -> readFully(table)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not exist in table schema"); + + masking.clear(); + masking.put( + "display", + new FieldTransform(new FieldRef(0, "_ROW_ID", DataTypes.BIGINT().notNull()))); + setColumnMasking(identifier, masking); + ReadBuilder readBuilder = table.newReadBuilder().withProjection(new int[] {0}); + assertThatThrownBy(() -> readBuilder.newScan().plan()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not project"); + } + + @Test + void testColumnMaskingDisablesFilterStatsPruning() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_filter_stats"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "amount", DataTypes.INT())); + fields.add(new DataField(1, "src", DataTypes.INT())); + Table table = + createMaskingAuthTable( + identifier, + fields, + Collections.singletonMap( + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b")); + for (int[] row : new int[][] {{1, 1000}, {900, 10}}) { + commitRows(table, GenericRow.of(row[0], row[1])); + } + setColumnMask( + identifier, "amount", new FieldTransform(new FieldRef(1, "src", DataTypes.INT()))); + + LeafPredicate amountFilter = leaf(0, "amount", DataTypes.INT(), GreaterThan.INSTANCE, 500); + ReadBuilder readBuilder = + table.newReadBuilder().withProjection(new int[] {0}).withFilter(amountFilter); + TableRead read = readBuilder.newRead(); + List rows = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("amount")); + assertThat( + rows.stream() + .map(row -> row.getInt(0)) + .collect(java.util.stream.Collectors.toList())) + .containsExactly(1000); + } + + @Test + void testStreamScanFilterOnMaskedColumnNotPushedToStats() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_stream_masking_filter_stats"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "amount", DataTypes.INT())); + fields.add(new DataField(1, "src", DataTypes.INT())); + Table table = + createMaskingAuthTable( + identifier, + fields, + Collections.singletonMap( + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b")); + for (int[] row : new int[][] {{1, 1000}, {900, 10}}) { + commitRows(table, GenericRow.of(row[0], row[1])); + } + setColumnMask( + identifier, "amount", new FieldTransform(new FieldRef(1, "src", DataTypes.INT()))); + + LeafPredicate amountFilter = leaf(0, "amount", DataTypes.INT(), GreaterThan.INSTANCE, 500); + ReadBuilder readBuilder = + table.newReadBuilder().withProjection(new int[] {0}).withFilter(amountFilter); + StreamTableScan scan = readBuilder.newStreamScan(); + List rows = + collectRows( + readBuilder.newRead().createReader(scan.plan().splits()), + table.rowType().project("amount")); + assertThat( + rows.stream() + .map(row -> row.getInt(0)) + .collect(java.util.stream.Collectors.toList())) + .containsExactly(1000); + } + + @Test + void testMaskGrowthOnPushedFilterColumn() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_pushed_filter"); + Table table = + createMaskingAuthTable( + identifier, stringFields("display", "other"), Collections.emptyMap()); + writeStringRow(table, "d1", "o1"); + + LeafPredicate displayFilter = + leaf( + 0, + "display", + DataTypes.STRING(), + Equal.INSTANCE, + BinaryString.fromString("d1")); + ReadBuilder readBuilder = + table.newReadBuilder().withProjection(new int[] {0}).withFilter(displayFilter); + TableScan scan = readBuilder.newScan(); + TableRead read = readBuilder.newRead(); + List plain = + collectRows( + read.createReader(scan.plan().splits()), + table.rowType().project("display")); + assertThat(plain).hasSize(1); + + maskConstant(identifier, "display"); + + assertThatThrownBy(scan::plan) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Recreate the scan"); + + List splits = readBuilder.newScan().plan().splits(); + List masked = + collectRows(read.createReader(splits), table.rowType().project("display")); + assertThat(masked).isEmpty(); + } + + @Test + void testDeferredFilterAppliesToPartitionListing() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "filter_partition_listing"); + catalog.createDatabase(identifier.getDatabaseName(), true); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "p", DataTypes.STRING())); + fields.add(new DataField(1, "v", DataTypes.STRING())); + catalog.createTable( + identifier, + new Schema( + fields, + Collections.singletonList("p"), + Collections.emptyList(), + Collections.emptyMap(), + ""), + true); + Table table = catalog.getTable(identifier); + writeStringRow(table, "a", "v1"); + writeStringRow(table, "b", "v2"); + + LeafPredicate partitionFilter = + leaf(0, "p", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("a")); + assertThat( + table.newReadBuilder() + .withFilter(partitionFilter) + .newScan() + .listPartitionEntries()) + .hasSize(1); + } + + @Test + void testRowFilterAppliesToPartitionListing() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_row_filter_partition_listing"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("p", "v"), + Collections.singletonList("p"), + Collections.emptyList(), + Collections.emptyMap()); + writeStringRows(table, new String[] {"a", "v1"}, new String[] {"b", "v2"}); + + LeafPredicate onA = + leaf(0, "p", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("a")); + setRowFilter(identifier, Collections.singletonList(onA)); + + List entries = + catalog.getTable(identifier).newReadBuilder().newScan().listPartitionEntries(); + assertThat(entries).hasSize(1); + assertThat(entries.get(0).partition().getString(0).toString()).isEqualTo("a"); + } + + @Test + void testColumnMaskingDisablesLimitPushdown() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_limit"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "amount", DataTypes.INT())); + fields.add(new DataField(1, "src", DataTypes.INT())); + Table table = + createMaskingAuthTable( + identifier, + fields, + Collections.singletonMap( + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b")); + for (int[] row : new int[][] {{900, 10}, {1, 1000}}) { + commitRows(table, GenericRow.of(row[0], row[1])); + } + setColumnMask( + identifier, "amount", new FieldTransform(new FieldRef(1, "src", DataTypes.INT()))); + + LeafPredicate amountFilter = leaf(0, "amount", DataTypes.INT(), GreaterThan.INSTANCE, 500); + ReadBuilder readBuilder = + table.newReadBuilder() + .withProjection(new int[] {0}) + .withFilter(amountFilter) + .withLimit(1); + TableRead read = readBuilder.newRead().executeFilter(); + List rows = + collectRows( + read.createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("amount")); + assertThat( + rows.stream() + .map(row -> row.getInt(0)) + .collect(java.util.stream.Collectors.toList())) + .contains(1000); + } + + @Test + void testFilterOnMaskedPartitionColumn() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_partition"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("p", "v"), + Collections.singletonList("p"), + Collections.emptyList(), + Collections.emptyMap()); + writeStringRow(table, "a", "va"); + writeStringRow(table, "b", "vb"); + + setColumnMask( + identifier, "p", new FieldTransform(new FieldRef(1, "v", DataTypes.STRING()))); + + LeafPredicate maskedMatch = + leaf(0, "p", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("vb")); + List rows = readWithFilter(table, maskedMatch, "p", "v"); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getString(0).toString()).isEqualTo("vb"); + assertThat(rows.get(0).getString(1).toString()).isEqualTo("vb"); + + LeafPredicate rawMatch = + leaf(0, "p", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("a")); + assertThat(readWithFilter(table, rawMatch, "p", "v")).isEmpty(); + + List unprojected = readWithFilter(table, maskedMatch, "v"); + assertThat(unprojected).hasSize(1); + assertThat(unprojected.get(0).getString(0).toString()).isEqualTo("vb"); + } + + @Test + void testLimitWithFilterOnMaskedPartitionColumn() throws Exception { + Identifier identifier = + Identifier.create("test_table_db", "auth_table_masking_partition_limit"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("p", "v"), + Collections.singletonList("p"), + Collections.emptyList(), + Collections.emptyMap()); + writeStringRow(table, "a", "va"); + writeStringRow(table, "b", "vb"); + setColumnMask( + identifier, "p", new FieldTransform(new FieldRef(1, "v", DataTypes.STRING()))); + + LeafPredicate maskedMatch = + leaf(0, "p", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("vb")); + ReadBuilder readBuilder = + table.newReadBuilder() + .withProjection(new int[] {0, 1}) + .withFilter(maskedMatch) + .withLimit(1); + List rows = + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + table.rowType().project("p", "v")); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getString(1).toString()).isEqualTo("vb"); + } + + @Test + void testMaskedPkFilterNotAppliedOnRawValues() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_table_masking_pk_filter"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "id", DataTypes.INT().notNull())); + fields.add(new DataField(1, "src", DataTypes.INT())); + Table table = + createMaskingAuthTable( + identifier, + fields, + Collections.emptyList(), + Collections.singletonList("id"), + Collections.singletonMap(CoreOptions.BUCKET.key(), "1")); + commitRows(table, GenericRow.of(1, 500), GenericRow.of(2, 600)); + + setColumnMask( + identifier, "id", new FieldTransform(new FieldRef(1, "src", DataTypes.INT()))); + + LeafPredicate idFilter = leaf(0, "id", DataTypes.INT(), Equal.INSTANCE, 500); + List rows = readWithFilter(table, idFilter, "id", "src"); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getInt(0)).isEqualTo(500); + } + + private List readWithFilter(Table table, Predicate filter, String... projected) + throws Exception { + int[] projection = table.rowType().getFieldIndices(Arrays.asList(projected)); + ReadBuilder readBuilder = + table.newReadBuilder().withProjection(projection).withFilter(filter); + return collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + table.rowType().project(projected)); + } + + private static void readFully(Table table) throws Exception { + ReadBuilder readBuilder = table.newReadBuilder(); + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + table.rowType()); + } + @Test void testRowFilter() throws Exception { Identifier identifier = Identifier.create("test_table_db", "auth_table_filter"); @@ -4342,6 +5520,482 @@ void testRowFilterReadLimitSkippedWithTopN() throws Exception { assertThat(result).containsExactlyInAnyOrder("+I[3, 30]", "+I[4, 40]"); } + @Test + void testColumnMaskingOrFilterWithUnprojectedOperand() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_or_filter_unprojected"); + Table table = + createMaskingAuthTable( + identifier, stringFields("a", "b", "c"), Collections.emptyMap()); + commitRows( + table, + GenericRow.of( + BinaryString.fromString("raw"), + BinaryString.fromString("bee"), + BinaryString.fromString("cee"))); + + setColumnMask( + identifier, + "a", + new ConcatTransform(Collections.singletonList(BinaryString.fromString("MASKED")))); + + Predicate onMasked = + leaf(0, "a", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("MASKED")); + Predicate onPlain = + leaf(1, "b", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("bee")); + Predicate disjunction = PredicateBuilder.or(onMasked, onPlain); + + ReadBuilder readBuilder = + table.newReadBuilder().withProjection(new int[] {2}).withFilter(disjunction); + List splits = readBuilder.newScan().plan().splits(); + List out = new ArrayList<>(); + try (RecordReader reader = readBuilder.newRead().createReader(splits)) { + reader.forEachRemaining(r -> out.add(r.getString(0).toString())); + } + assertThat(out).containsExactly("cee"); + } + + @Test + void testColumnMaskingPartitionListingBeforePlan() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_partition_listing"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("pt", "a"), + Collections.singletonList("pt"), + Collections.emptyList(), + Collections.emptyMap()); + commitRows( + table, + GenericRow.of(BinaryString.fromString("p1"), BinaryString.fromString("raw"))); + + setColumnMask( + identifier, + "a", + new ConcatTransform(Collections.singletonList(BinaryString.fromString("MASKED")))); + + Predicate onMasked = + leaf(1, "a", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("MASKED")); + + InnerTableScan scan = + (InnerTableScan) table.newReadBuilder().withFilter(onMasked).newScan(); + assertThat(scan.listPartitionEntries()).hasSize(1); + assertThat(scan.plan().splits()).isNotEmpty(); + } + + @Test + void testQueryAuthLimitDoesNotCutRowsBeforeFilter() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_limit_with_filter"); + Table table = + createMaskingAuthTable(identifier, stringFields("a", "b"), Collections.emptyMap()); + commitRows( + table, + GenericRow.of(BinaryString.fromString("no"), BinaryString.fromString("r1")), + GenericRow.of(BinaryString.fromString("yes"), BinaryString.fromString("r2"))); + + Predicate onA = + leaf(0, "a", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("yes")); + ReadBuilder readBuilder = table.newReadBuilder().withFilter(onA).withLimit(1); + List out = new ArrayList<>(); + try (RecordReader reader = + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits())) { + reader.forEachRemaining(r -> out.add(r.getString(1).toString())); + } + assertThat(out).contains("r2"); + } + + @Test + void testMaskedPartitionKeyRejectsPartitionFilter() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_masked_part_filter"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("p", "v"), + Collections.singletonList("p"), + Collections.emptyList(), + Collections.emptyMap()); + writeStringRow(table, "a", "va"); + writeStringRow(table, "b", "vb"); + + setColumnMask( + identifier, "p", new FieldTransform(new FieldRef(1, "v", DataTypes.STRING()))); + + LeafPredicate maskedMatch = + leaf(0, "p", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("vb")); + + InnerTableScan partitionScan = (InnerTableScan) table.newReadBuilder().newScan(); + partitionScan.withPartitionFilter(maskedMatch); + assertThatThrownBy(partitionScan::plan) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("masked partition key"); + + assertThat(readWithFilter(table, maskedMatch, "p", "v")).hasSize(1); + } + + @Test + void testPartitionFilterAllowedOnUnmaskedPartitionKey() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_two_part_keys"); + 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"}); + + setColumnMask( + identifier, "p2", new FieldTransform(new FieldRef(2, "v", DataTypes.STRING()))); + + LeafPredicate onP1 = + leaf(0, "p1", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("x")); + InnerTableScan okScan = (InnerTableScan) table.newReadBuilder().newScan(); + okScan.withPartitionFilter(onP1); + assertThat(okScan.plan().splits()).isNotEmpty(); + + LeafPredicate onP2 = + leaf(1, "p2", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("v1")); + InnerTableScan badScan = (InnerTableScan) table.newReadBuilder().newScan(); + badScan.withPartitionFilter(onP2); + assertThatThrownBy(badScan::plan) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("masked partition key"); + } + + @Test + void testExplicitPartitionFilterSurvivesDeferredQueryFilter() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_part_filter_with_filter"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("p", "v"), + Collections.singletonList("p"), + Collections.emptyList(), + Collections.emptyMap()); + writeStringRows(table, new String[] {"a", "va"}, new String[] {"b", "vb"}); + + LeafPredicate pAtLeastA = + leaf( + 0, + "p", + DataTypes.STRING(), + GreaterOrEqual.INSTANCE, + BinaryString.fromString("a")); + assertThat(readPartitionB(table, pAtLeastA)).containsExactly("b"); + + Identifier plain = Identifier.create("test_table_db", "plain_part_filter_with_filter"); + catalog.createTable( + plain, + new Schema( + stringFields("p", "v"), + Collections.singletonList("p"), + Collections.emptyList(), + Collections.emptyMap(), + ""), + true); + Table plainTable = catalog.getTable(plain); + writeStringRows(plainTable, new String[] {"a", "va"}, new String[] {"b", "vb"}); + assertThat(readPartitionB(plainTable, pAtLeastA)).containsExactly("b"); + } + + private static List readPartitionB(Table table, Predicate filter) throws Exception { + ReadBuilder readBuilder = + table.newReadBuilder() + .withFilter(filter) + .withPartitionFilter(Collections.singletonMap("p", "b")); + return collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + table.rowType()) + .stream() + .map(row -> row.getString(0).toString()) + .distinct() + .collect(Collectors.toList()); + } + + @Test + void testQueryAuthLimitAppliesWhenReaderExecutesFilter() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_limit_execute_filter"); + Table table = + createMaskingAuthTable(identifier, stringFields("a", "b"), Collections.emptyMap()); + commitRows( + table, + GenericRow.of(BinaryString.fromString("yes"), BinaryString.fromString("r1")), + GenericRow.of(BinaryString.fromString("yes"), BinaryString.fromString("r2")), + GenericRow.of(BinaryString.fromString("yes"), BinaryString.fromString("r3"))); + + Predicate onA = + leaf(0, "a", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("yes")); + ReadBuilder readBuilder = table.newReadBuilder().withFilter(onA).withLimit(2); + List out = new ArrayList<>(); + try (RecordReader reader = + readBuilder + .newRead() + .executeFilter() + .createReader(readBuilder.newScan().plan().splits())) { + reader.forEachRemaining(r -> out.add(r.getString(1).toString())); + } + assertThat(out).hasSize(2); + } + + @Test + void testPartitionFilterFieldsReplacedOnReapply() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_part_filter_reapply"); + Table table = + createMaskingAuthTable( + identifier, + stringFields("p1", "p2", "v"), + Arrays.asList("p1", "p2"), + Collections.emptyList(), + Collections.emptyMap()); + writeStringRows(table, new String[] {"x", "a", "v1"}); + + setColumnMask( + identifier, "p2", new FieldTransform(new FieldRef(2, "v", DataTypes.STRING()))); + + LeafPredicate onMaskedP2 = + leaf(1, "p2", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("v1")); + LeafPredicate onPlainP1 = + leaf(0, "p1", DataTypes.STRING(), Equal.INSTANCE, BinaryString.fromString("x")); + + InnerTableScan scan = (InnerTableScan) table.newReadBuilder().newScan(); + scan.withPartitionFilter(onMaskedP2); + scan.withPartitionFilter(onPlainP1); + assertThat(scan.plan().splits()).isNotEmpty(); + } + + @Test + void testPhysicalMetadataSystemTablesRejectedUnderQueryAuth() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_systab_physical"); + Table table = + createMaskingAuthTable( + identifier, stringFields("secret", "other"), Collections.emptyMap()); + writeStringRow(table, "TOPSECRET", "o1"); + maskConstant(identifier, "secret"); + + for (String suffix : Arrays.asList("files", "file_key_ranges", "binlog", "statistics")) { + Identifier sysId = + Identifier.create( + identifier.getDatabaseName(), + identifier.getObjectName() + "$" + suffix); + assertThatThrownBy(() -> catalog.getTable(sysId)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("query-auth table"); + } + + for (String suffix : Arrays.asList("audit_log", "ro")) { + Identifier sysId = + Identifier.create( + identifier.getDatabaseName(), + identifier.getObjectName() + "$" + suffix); + assertThat(batchRead(catalog.getTable(sysId)).toString()) + .contains("****") + .doesNotContain("TOPSECRET"); + } + } + + @Test + void testMaskReadingAnotherMaskedColumnRejected() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_mask_compose"); + Table table = + createMaskingAuthTable( + identifier, stringFields("secret", "display"), Collections.emptyMap()); + writeStringRow(table, "TOPSECRET", "ignored"); + + Map compose = new HashMap<>(); + compose.put( + "secret", + new ConcatTransform(Collections.singletonList(BinaryString.fromString("****")))); + compose.put("display", new FieldTransform(new FieldRef(0, "secret", DataTypes.STRING()))); + setColumnMasking(identifier, compose); + assertThatThrownBy(() -> batchRead(table)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("which is masked too"); + + Map plain = new HashMap<>(); + plain.put("display", new FieldTransform(new FieldRef(0, "secret", DataTypes.STRING()))); + setColumnMasking(identifier, plain); + assertThat(batchRead(table).toString()).contains("TOPSECRET"); + + Map selfRef = new HashMap<>(); + selfRef.put( + "secret", + new UpperTransform( + Collections.singletonList(new FieldRef(0, "secret", DataTypes.STRING())))); + setColumnMasking(identifier, selfRef); + assertThat(batchRead(table).toString()).contains("TOPSECRET"); + } + + @Test + void testQueryAuthRejectedWhereItCannotBeEnforced() throws Exception { + for (String type : Arrays.asList("format-table", "object-table")) { + Map opts = new HashMap<>(); + opts.put(QUERY_AUTH_ENABLED.key(), "true"); + opts.put("type", type); + opts.put("file.format", "csv"); + Identifier id = + Identifier.create( + "test_table_db", "auth_unsupported_" + type.replace('-', '_')); + assertThatThrownBy( + () -> + catalog.createTable( + id, + new Schema( + stringFields("secret"), + Collections.emptyList(), + Collections.emptyList(), + opts, + ""), + false)) + .hasMessageContaining(QUERY_AUTH_ENABLED.key()); + } + + Identifier identifier = Identifier.create("test_table_db", "auth_search_rejected"); + Table table = + createMaskingAuthTable( + identifier, stringFields("secret", "other"), Collections.emptyMap()); + assertThatThrownBy( + () -> ((FileStoreTable) table).newFullTextSearchBuilder().newFullTextScan()) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("query-auth table"); + + assertThatThrownBy(() -> new LocalTableQuery((FileStoreTable) table)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("query-auth table"); + + assertThatThrownBy(() -> ((FileStoreTable) table).newVectorSearchBuilder().newVectorRead()) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("query-auth table"); + } + + @Test + void testRowIdFilterOnDataEvolutionQueryAuthTable() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_de_rowid_filter"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "f0", DataTypes.INT())); + fields.add(new DataField(1, "f1", DataTypes.STRING())); + Map options = new HashMap<>(); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + Table table = createMaskingAuthTable(identifier, fields, options); + + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite()) { + write.write(GenericRow.of(0, BinaryString.fromString("a0"))); + write.write(GenericRow.of(1, BinaryString.fromString("a1"))); + builder.newCommit().commit(write.prepareCommit()); + } + + Predicate onRowId = + LeafPredicate.of( + new FieldTransform( + new FieldRef( + SpecialFields.ROW_ID.id(), + SpecialFields.ROW_ID.name(), + DataTypes.BIGINT())), + Equal.INSTANCE, + Collections.singletonList(0L)); + ReadBuilder readBuilder = table.newReadBuilder().withFilter(onRowId); + assertThat(readBuilder.newScan().plan().splits()).isNotEmpty(); + } + + @Test + void testSuppliedGlobalIndexResultIgnoredUnderQueryAuth() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_de_supplied_index"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "f0", DataTypes.INT())); + Map options = new HashMap<>(); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + options.put(CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b"); + Table table = createMaskingAuthTable(identifier, fields, options); + + for (int v : new int[] {10, 20, 30}) { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite()) { + write.write(GenericRow.of(v)); + builder.newCommit().commit(write.prepareCommit()); + } + } + + InnerTableScan scan = (InnerTableScan) table.newReadBuilder().newScan(); + scan.withGlobalIndexResult(GlobalIndexResult.fromRange(new Range(0, 1))); + assertThat(scan.plan().splits()).hasSize(3); + } + + @Test + void testRowIdFilterWithLimitOnDataEvolutionQueryAuthTable() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_de_rowid_limit"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "f0", DataTypes.INT())); + Map options = new HashMap<>(); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + options.put(CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b"); + Table table = createMaskingAuthTable(identifier, fields, options); + + for (int v : new int[] {10, 20, 30}) { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite()) { + write.write(GenericRow.of(v)); + builder.newCommit().commit(write.prepareCommit()); + } + } + + Predicate onLastRowId = + LeafPredicate.of( + new FieldTransform( + new FieldRef( + SpecialFields.ROW_ID.id(), + SpecialFields.ROW_ID.name(), + DataTypes.BIGINT())), + Equal.INSTANCE, + Collections.singletonList(2L)); + ReadBuilder readBuilder = table.newReadBuilder().withFilter(onLastRowId).withLimit(1); + assertThat(readBuilder.newScan().plan().splits()).hasSize(3); + } + + @Test + void testMaskedRowIdFilterOnDataEvolutionQueryAuthTable() throws Exception { + Identifier identifier = Identifier.create("test_table_db", "auth_de_masked_rowid"); + List fields = new ArrayList<>(); + fields.add(new DataField(0, "f0", DataTypes.BIGINT())); + Map options = new HashMap<>(); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + Table table = createMaskingAuthTable(identifier, fields, options); + + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite()) { + write.write(GenericRow.of(42L)); + builder.newCommit().commit(write.prepareCommit()); + } + + Map masking = new HashMap<>(); + masking.put( + SpecialFields.ROW_ID.name(), + new FieldTransform(new FieldRef(0, "f0", DataTypes.BIGINT()))); + setColumnMasking(identifier, masking); + + Predicate onMaskedRowId = + LeafPredicate.of( + new FieldTransform( + new FieldRef( + SpecialFields.ROW_ID.id(), + SpecialFields.ROW_ID.name(), + DataTypes.BIGINT())), + Equal.INSTANCE, + Collections.singletonList(42L)); + RowType readType = + new RowType(Arrays.asList(table.rowType().getField("f0"), SpecialFields.ROW_ID)); + ReadBuilder readBuilder = + table.newReadBuilder().withReadType(readType).withFilter(onMaskedRowId); + List rows = + collectRows( + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits()), + readType); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(1)).isEqualTo(42L); + } + @Test void testRowFilterWithTopNKeepsAuthorizedSplits() throws Exception { Identifier identifier = Identifier.create("test_table_db", "auth_table_topn"); diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java index 5f5aa2fce0eb..b7da752f7b06 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java @@ -769,6 +769,35 @@ mergeOptions, singletonList("id"), emptyList()))) "Primary-key managed BLOB tables do not support 'pk-clustering-override'."); } + @Test + public void testQueryAuthOnlyOnFileStoreTables() { + for (String type : Arrays.asList("format-table", "object-table")) { + Map options = new HashMap<>(); + options.put(CoreOptions.QUERY_AUTH_ENABLED.key(), "true"); + options.put(CoreOptions.TYPE.key(), type); + assertThatThrownBy(() -> validateTableSchema(queryAuthSchema(options))) + .hasMessageContaining(CoreOptions.QUERY_AUTH_ENABLED.key()); + } + + for (String type : Arrays.asList("table", "materialized-table")) { + Map options = new HashMap<>(); + options.put(CoreOptions.QUERY_AUTH_ENABLED.key(), "true"); + options.put(CoreOptions.TYPE.key(), type); + validateTableSchema(queryAuthSchema(options)); + } + } + + private TableSchema queryAuthSchema(Map options) { + return new TableSchema( + 1, + singletonList(new DataField(0, "id", DataTypes.INT())), + 10, + emptyList(), + emptyList(), + options, + ""); + } + private TableSchema primaryKeyBlobSchema( Map options, List primaryKeys, List partitionKeys) { return new TableSchema( diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupDataTableScan.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupDataTableScan.java index b64e7edace97..7356c670f351 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupDataTableScan.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupDataTableScan.java @@ -64,6 +64,7 @@ public LookupDataTableScan( LookupStreamScanMode lookupScanMode) { super( table.schema(), + table.schemaManager(), table.coreOptions(), snapshotReader, table.snapshotManager(), diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkVectorSearchBuilderImpl.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkVectorSearchBuilderImpl.java index cad0f8683536..d4bf9748634f 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkVectorSearchBuilderImpl.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkVectorSearchBuilderImpl.java @@ -40,6 +40,7 @@ public FlinkVectorSearchBuilderImpl(InnerTable table, StreamExecutionEnvironment @Override public VectorRead newVectorRead() { + rejectUnderQueryAuth(); checkNotNull(vector, "vector must be set via withVector()"); if (isPrimaryKeyVectorSearch()) { return new FlinkPrimaryKeyVectorRead( diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java index 5be454a9d0c8..6e606d1a2d7f 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java @@ -365,6 +365,85 @@ public void testColumnMasking() { Row.of("user1@example.com"), Row.of("user2@example.com")); } + @Test + public void testColumnMaskingCrossColumnWithProjection() { + String maskingTable = "cross_column_masking_table"; + batchSql( + String.format( + "CREATE TABLE %s.%s (first_name STRING, last_name STRING, display STRING, other_col STRING)" + + " WITH ('query-auth.enabled' = 'true', 'source.split.target-size' = '1 b')", + DATABASE_NAME, maskingTable)); + batchSql( + String.format( + "INSERT INTO %s.%s VALUES ('john', 'doe', 'ignored', 'o1')", + DATABASE_NAME, maskingTable)); + batchSql( + String.format( + "INSERT INTO %s.%s VALUES ('jane', 'roe', 'ignored', 'o2')", + DATABASE_NAME, maskingTable)); + + Map columnMasking = new HashMap<>(); + columnMasking.put( + "display", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(0, "first_name", DataTypes.STRING()), + new FieldRef(1, "last_name", DataTypes.STRING())))); + restCatalogServer.setColumnMaskingAuth( + Identifier.create(DATABASE_NAME, maskingTable), columnMasking); + + assertThat( + batchSql( + String.format( + "SELECT display FROM %s.%s", DATABASE_NAME, maskingTable))) + .containsExactlyInAnyOrder(Row.of("john-doe"), Row.of("jane-roe")); + assertThat( + batchSql( + String.format( + "SELECT other_col FROM %s.%s", + DATABASE_NAME, maskingTable))) + .containsExactlyInAnyOrder(Row.of("o1"), Row.of("o2")); + } + + @Test + public void testFilterOnMaskedPartitionColumn() { + String maskingTable = "partition_masking_table"; + batchSql( + String.format( + "CREATE TABLE %s.%s (p STRING, v STRING) PARTITIONED BY (p)" + + " WITH ('query-auth.enabled' = 'true')", + DATABASE_NAME, maskingTable)); + batchSql( + String.format( + "INSERT INTO %s.%s VALUES ('a', 'va'), ('b', 'vb')", + DATABASE_NAME, maskingTable)); + + Map columnMasking = new HashMap<>(); + columnMasking.put("p", new FieldTransform(new FieldRef(1, "v", DataTypes.STRING()))); + restCatalogServer.setColumnMaskingAuth( + Identifier.create(DATABASE_NAME, maskingTable), columnMasking); + + assertThat( + batchSql( + String.format( + "SELECT p, v FROM %s.%s WHERE p = 'vb'", + DATABASE_NAME, maskingTable))) + .containsExactlyInAnyOrder(Row.of("vb", "vb")); + assertThat( + batchSql( + String.format( + "SELECT p, v FROM %s.%s WHERE p = 'a'", + DATABASE_NAME, maskingTable))) + .isEmpty(); + assertThat( + batchSql( + String.format( + "SELECT v FROM %s.%s WHERE p = 'vb'", + DATABASE_NAME, maskingTable))) + .containsExactlyInAnyOrder(Row.of("vb")); + } + @Test public void testRowFilter() { String filterTable = "row_filter_table"; @@ -758,7 +837,8 @@ public void testColumnMaskingAndRowFilter() { "SELECT id, name FROM %s.%s WHERE age > 30 ORDER BY id", DATABASE_NAME, combinedTable))) .rootCause() - .hasMessageContaining("Unable to read data without column non_existent_column"); + .hasMessageContaining( + "Row filter references column 'non_existent_column' which does not exist"); // Clear both column masking and row filter restCatalogServer.setColumnMaskingAuth( diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java index a1748b6513cf..3c8eedc3a580 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java @@ -38,6 +38,7 @@ public SparkVectorSearchBuilderImpl(InnerTable table) { @Override public VectorRead newVectorRead() { + rejectUnderQueryAuth(); if (isPrimaryKeyVectorSearch()) { return new SparkPrimaryKeyVectorRead( table, vectorColumn, vector, limit, options, filter); diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java index fcde06202ed0..c9ff2b5fd34a 100644 --- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java +++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java @@ -364,6 +364,56 @@ public void testColumnMasking() { .isEqualTo("[[user1@example.com], [user2@example.com]]"); } + @Test + public void testColumnMaskingCrossColumnWithProjection() { + spark.sql( + "CREATE TABLE t_cross_column_masking (first_name STRING, last_name STRING, display STRING, other_col STRING)" + + " TBLPROPERTIES ('query-auth.enabled'='true', 'source.split.target-size'='1 b')"); + spark.sql("INSERT INTO t_cross_column_masking VALUES ('john', 'doe', 'ignored', 'o1')"); + spark.sql("INSERT INTO t_cross_column_masking VALUES ('jane', 'roe', 'ignored', 'o2')"); + + Map columnMasking = new HashMap<>(); + columnMasking.put( + "display", + new ConcatWsTransform( + Arrays.asList( + BinaryString.fromString("-"), + new FieldRef(0, "first_name", DataTypes.STRING()), + new FieldRef(1, "last_name", DataTypes.STRING())))); + restCatalogServer.setColumnMaskingAuth( + Identifier.create("db2", "t_cross_column_masking"), columnMasking); + + assertThat( + spark.sql("SELECT display FROM t_cross_column_masking ORDER BY other_col") + .collectAsList() + .toString()) + .isEqualTo("[[john-doe], [jane-roe]]"); + assertThat( + spark.sql("SELECT other_col FROM t_cross_column_masking ORDER BY other_col") + .collectAsList() + .toString()) + .isEqualTo("[[o1], [o2]]"); + } + + @Test + public void testRowFilterDisablesAggregatePushdown() { + spark.sql( + "CREATE TABLE t_agg_pushdown (id INT) TBLPROPERTIES" + + " ('query-auth.enabled'='true')"); + spark.sql("INSERT INTO t_agg_pushdown VALUES (1), (2), (3)"); + + LeafPredicate idFilter = + LeafPredicate.of( + new FieldTransform(new FieldRef(0, "id", DataTypes.INT())), + GreaterThan.INSTANCE, + Collections.singletonList(1)); + restCatalogServer.setRowFilterAuth( + Identifier.create("db2", "t_agg_pushdown"), Collections.singletonList(idFilter)); + + assertThat(spark.sql("SELECT COUNT(*) FROM t_agg_pushdown").collectAsList().toString()) + .isEqualTo("[[2]]"); + } + @Test public void testRowFilter() { spark.sql( @@ -863,7 +913,8 @@ public void testColumnMaskingAndRowFilter() { spark.sql( "SELECT id, name FROM t_combined WHERE age > 30 ORDER BY id") .collectAsList()) - .hasMessageContaining("Unable to read data without column non_existent_column"); + .hasMessageContaining( + "Row filter references column 'non_existent_column' which does not exist"); // Clear both column masking and row filter restCatalogServer.setColumnMaskingAuth(