OIiveirra commented on code in PR #68161: URL: https://github.com/apache/doris/pull/68161#discussion_r4302890790
########## 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: Confirmed and fixed in bc433ca6d8b48cc200fddd501eadfc6e4a9ce7df. Policy-field discovery and augmentation now use the schema of the branch selected by the split, following the same branch ordering/dispatch as the SDK and reusing the existing ordering resolver. SQL output projection and policy expressions are unchanged. Added real historical Parquet reader tests with an independently renamed hidden integer policy column under both default and primary ordering; they verify policy exclusion and repaired positive/negative timestamp values. The default-order case failed before the fix. All 78 focused JNI tests and Maven Checkstyle passed after the fix. -- 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]
