github-actions[bot] commented on code in PR #68161: URL: https://github.com/apache/doris/pull/68161#discussion_r4293498265
########## fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonTimestampQueryAuth.java: ########## @@ -0,0 +1,204 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.paimon; + +import org.apache.paimon.catalog.TableQueryAuthResult; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericMap; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalArray; +import org.apache.paimon.data.InternalMap; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.Timestamp; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateVisitor; +import org.apache.paimon.predicate.Transform; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.table.FallbackReadFileStoreTable.FallbackSplitImpl; +import org.apache.paimon.table.source.QueryAuthSplit; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeChecks; +import org.apache.paimon.types.LocalZonedTimestampType; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TimestampType; +import org.apache.paimon.utils.InternalRowUtils; + +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +/** Keeps SDK authorization and masking after current-precision materialization. */ +final class PaimonTimestampQueryAuth extends TableQueryAuthResult { + private static final long serialVersionUID = 1L; + private final RowType currentReadType; + private final TableQueryAuthResult original; + + PaimonTimestampQueryAuth(TableQueryAuthResult original, RowType currentReadType) { + super(original.filter(), original.columnMasking()); + this.original = original; + this.currentReadType = currentReadType; + } + + @Override + public Predicate extractPredicate() { + return original.extractPredicate(); + } + + @Override + public Map<String, Transform> extractColumnMasking() { + return original.extractColumnMasking(); + } + + static QueryAuthSplit findAuthSplit(Split split) { + // FallbackDataSplit is a leaf whose wrapped() returns itself, not a split container. + if (split instanceof FallbackSplitImpl) { + return findAuthSplit(((FallbackSplitImpl) split).wrapped()); + } + return split instanceof QueryAuthSplit ? (QueryAuthSplit) split : null; + } + + static Split repairPolicy(Split split, RowType currentReadType) { + if (split instanceof FallbackSplitImpl) { + FallbackSplitImpl fallback = (FallbackSplitImpl) split; + return new FallbackSplitImpl(repairPolicy(fallback.wrapped(), currentReadType), fallback.isFallback()); + } + QueryAuthSplit auth = (QueryAuthSplit) split; + return new QueryAuthSplit(auth.split(), new PaimonTimestampQueryAuth(auth.authResult(), currentReadType)); + } + + static void addPolicyFields(TableQueryAuthResult auth, RowType tableType, + List<DataField> readFields, List<DataField> currentFields) { + List<String> projected = readFields.stream().map(DataField::name).collect(Collectors.toList()); + Set<String> required = requiredFields(auth, new LinkedHashSet<>(projected)); + for (DataField field : tableType.getFields()) { + if (required.contains(field.name())) { + int index = projected.indexOf(field.name()); + DataField widened = field.newType(PaimonJniScanner.createSafeTimestampReadType(field.type())); + if (index < 0) { + currentFields.add(field); + readFields.add(widened); + } else if (field.type() instanceof RowType || field.type() instanceof ArrayType + || field.type() instanceof MapType) { + // A policy can read a sibling that SQL pruned from a projected complex column. + currentFields.set(index, field); + readFields.set(index, widened); + } + } + } + } + + static boolean readsTimestamp(TableQueryAuthResult auth, RowType tableType, List<String> projected) { + Set<String> required = requiredFields(auth, new LinkedHashSet<>(projected)); + return tableType.getFields().stream().anyMatch(field -> required.contains(field.name()) + && PaimonJniScanner.containsTimestampType(field.type())); + } + + private static Set<String> requiredFields(TableQueryAuthResult auth, Set<String> projected) { + Set<String> required = new LinkedHashSet<>(); + Predicate filter = auth.extractPredicate(); + if (filter != null) { + required.addAll(PredicateVisitor.collectFieldNames(filter)); + } + for (Map.Entry<String, Transform> mask : auth.extractColumnMasking().entrySet()) { + if (!projected.contains(mask.getKey())) { + continue; + } + required.add(mask.getKey()); + for (Object input : mask.getValue().inputs()) { + if (input instanceof FieldRef) { + required.add(((FieldRef) input).name()); + } + } + } + return required; + } + + static RecordReader<InternalRow> projectOutput(RecordReader<InternalRow> reader, + RowType currentReadType, RowType outputType) { + if (currentReadType.equals(outputType)) { + return reader; + } + return reader.transform(row -> repairRow(row, currentReadType, outputType)); + } + + @Override + public RecordReader<InternalRow> doAuth(RecordReader<InternalRow> reader, RowType readType) { + // Delegate the policy itself to Paimon; neither filters nor masks are dropped or rewritten. + // System-table readers synthesize fields such as rowkind only after this inner callback. + RowType currentDataType = currentReadType.project(readType.getFieldNames()); Review Comment: [P1] Use the fallback branch schema for hidden policy fields Paimon [accepts fallback schemas whose positional types agree](https://github.com/apache/paimon/blob/release-1.4.2/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java#L211-L241), even when a nonpartition field is renamed. With main `(pt INT, id INT, ts TIMESTAMP(0))`, fallback `(pt INT, renamed_id INT, ts TIMESTAMP(0))`, historical p6 fallback files, and query auth enabled only on fallback with the serialized integer policy `renamed_id = 7`, `SELECT ts` from a fallback-only partition reaches this adapter. `addPolicyFields` uses the main `table.rowType()` and omits `renamed_id`; the [fallback SDK then adds that hidden field](https://github.com/apache/paimon/blob/release-1.4.2/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableRead.java#L112-L140) before invoking `doAuth`. This projection looks up `renamed_id` in a `currentReadType` containing only `ts`, indexes -1, and fails reader creation. The base SDK path could enforce this i nteger policy and return the positive p6-to-p0 timestamp. Resolve policy fields/current precision from the branch selected by the split and cover an independently renamed hidden policy column. This is distinct from the earlier fallback-wrapper thread: the adapter is reached, but its target schema belongs to the other branch. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
