[core] Fix column masking correctness in query-auth reads

paimon-common/src/main/java/org/apache/paimon/predicate/PredicateVisitor.java

@@ -40,6 +40,16 @@ static Set<String> collectFieldNames(@Nullable Predicate predicate) {
  return predicate.visit(new FieldNameCollector());
  }
 
+ static Set<String> collectTransformFieldNames(Transform transform) {
+ Set<String> fieldNames = new HashSet<>();
+ for (Object input : transform.inputs()) {
+ if (input instanceof FieldRef) {
+ fieldNames.add(((FieldRef) input).name());
+ }
+ }
+ return fieldNames;
+ }
+
  static Set<Integer> collectFieldIds(RowType rowType, @Nullable Predicate predicate) {
  if (predicate == null) {
  return Collections.emptySet();
@@ -58,13 +68,7 @@ class FieldNameCollector implements PredicateVisitor<Set<String>> {
 
  @Override
  public Set<String> visit(LeafPredicate predicate) {
- Set<String> 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

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());
  }

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<List<Range>> 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<List<Range>> 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<GlobalIndexResult> 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);
  }

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<DataField> 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;
  }

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<KeyValue> 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();
  }
 

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -156,6 +156,10 @@ public static void validateTableSchema(TableSchema schema, Set<String> 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<String> 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());

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(),

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)) {

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<PartitionEntry> 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<StartingScanner.Result> 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();