OIiveirra commented on code in PR #68161:
URL: https://github.com/apache/doris/pull/68161#discussion_r4309864905
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -2087,27 +2254,61 @@ static FileStoreTable
legacyOrcSplitTable(FileStoreTable table, DataSplit split)
return base;
}
- static boolean requiresLegacyOrcTimestampReader(Table table,
Optional<List<RawFile>> rawFiles,
+ private static boolean requiresCrossFamilyTimestampReader(FileStoreTable
table, Optional<List<RawFile>> rawFiles,
+ Set<Integer> readFieldIds, Map<Long, Boolean> schemaTimestamps) {
+ RowType readType = new RowType(table.rowType().getFields().stream()
+ .filter(field ->
readFieldIds.contains(field.id())).collect(Collectors.toList()));
+ if (!containsTimestampType(readType)) {
+ return false;
+ }
+ for (RawFile file : rawFiles.get()) {
+ if (schemaTimestamps.computeIfAbsent(file.schemaId(), id ->
hasTimestampPrecisionNarrowing(
+ table.schemaManager().schema(id).logicalRowType(),
readType, 0, true))) {
+ // Generic native casts round after zone conversion. The SDK
keeps its conversion
+ // semantics while the JNI widened read type delays truncation
until materialization.
+ return true;
+ }
+ }
+ return false;
+ }
+
+ static boolean requiresOrcTimestampReader(Table table,
Optional<List<RawFile>> rawFiles,
Set<Integer> readFieldIds, Map<Long, Boolean> schemaTimestamps) {
if (readFieldIds.isEmpty() || !rawFiles.isPresent()
- || rawFiles.get().stream().noneMatch(f ->
f.path().endsWith(".orc"))
- || !new org.apache.paimon.options.Options(table.options()).get(
-
org.apache.paimon.format.OrcOptions.ORC_TIMESTAMP_LTZ_LEGACY_TYPE)) {
+ || rawFiles.get().stream().noneMatch(f ->
f.path().endsWith(".orc"))) {
return false;
}
// Only fields present in a file require SDK timezone conversion. A
newly added nullable LTZ
// column reads as NULL in older files and must not disable their
native metadata scans.
FileStoreTable fileStoreTable = (FileStoreTable) table;
+ boolean legacyLtz = new
org.apache.paimon.options.Options(table.options()).get(
+
org.apache.paimon.format.OrcOptions.ORC_TIMESTAMP_LTZ_LEGACY_TYPE);
+ RowType readType = new
RowType(fileStoreTable.rowType().getFields().stream()
+ .filter(field ->
readFieldIds.contains(field.id())).collect(Collectors.toList()));
Review Comment:
Fixed in a6d3bedf4ca5c290ce6d6a1d6cbf448f4b6ec8e5. The scan selection is now
translated from outer-table field IDs to the selected branch's IDs by
positional correspondence, before both ORC and cross-family timestamp-history
routing. This does not assume matching IDs or names. Tests cover independently
evolved IDs and renamed fields, both branch orderings, filter-only timestamps,
and unaffected integer-only native scans. The new routing test failed before
the fix and passes afterwards. All 216 focused FE/JNI tests and the personal
paimon_timestamp_types external suite passed; ASAN build, connector packaging
and Checkstyle also passed. No V1 or ordinary SQL CAST changes are included.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -1241,15 +1251,27 @@ public Map<String, String> getScanNodeProperties(
// ("encodedStr is null"). Mirrors legacy
PaimonScanNode.createScanRangeLocations, which always called
// setPaimonPredicate(encodeObjectToString(predicates)) regardless of
whether predicates was empty.
List<org.apache.paimon.predicate.Predicate> predicates =
Collections.emptyList();
+ // The same predicate must stay out of the BE Paimon reader: JNI
materializes and repairs the
+ // value first, then the generic scan path evaluates the residual
against the repaired column.
if (filter.isPresent()) {
RowType rowType = table.rowType();
- PaimonPredicateConverter converter = new
PaimonPredicateConverter(rowType);
+ PaimonPredicateConverter converter = new PaimonPredicateConverter(
+ rowType, filterReadsTimestamp(rowType, filter.get())
+ && requiresTimestampPrecisionRepair(session,
table, paimonHandle));
predicates = converter.convert(filter.get());
}
props.put("paimon.predicate", encodeObjectToString(predicates));
// Paimon JDBC metastore options for BE (if applicable)
Map<String, String> backendOptions = new
LinkedHashMap<>(getBackendPaimonOptions());
+ Set<Integer> readFieldIds = scanReadFieldIds(table.rowType(), columns,
filter);
+ boolean readsTimestamp =
table.rowType().getFields().stream().anyMatch(field ->
+ readFieldIds.contains(field.id())
+ && containsTimestampType(field.type()));
+ backendOptions.put(DORIS_TIMESTAMP_PRECISION_REPAIR, String.valueOf(
+ (readsTimestamp || new
org.apache.paimon.options.Options(table.options()).get(
Review Comment:
Fixed in a6d3bedf4ca5c290ce6d6a1d6cbf448f4b6ec8e5. FE now checks query-auth
enablement across the pinned fallback branches, including the existing
supported system-table source resolution. The repair flag still requires
historical timestamp narrowing; the JNI explicit-false contract and
authenticated, statement-scoped history lookup are preserved. Added FE coverage
with auth enabled exclusively on the secondary branch and integer-only output,
plus no-history controls. A historical JNI reader test covers the hidden
p6-to-p4 .000001 typed SDK policy. The FE flag test failed before the fix. All
216 focused FE/JNI tests and the personal timestamp external suite passed. This
does not claim end-to-end coverage of remote JSON timestamp-policy transport.
##########
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:
Follow-up fixed in a6d3bedf4ca5c290ce6d6a1d6cbf448f4b6ec8e5: the
selected-branch policy-schema lookup now also traverses row-backed system
wrappers using the existing pinned-source/delegate helpers, without reopening
the catalog. The audit-log outer reader still synthesizes rowkind; policy
filters and masks are unchanged. Added historical audit-log/fallback reader
coverage with an independently renamed hidden integer policy column, rowkind
plus ts output, positive and negative-epoch values, policy exclusion, and both
branch orderings. The default-order audit case failed before the fix. All 216
focused FE/JNI tests and the personal paimon_timestamp_types external suite
passed. This supplements the direct-fallback fix previously recorded in this
resolved thread.
--
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]
