JingsongLi commented on code in PR #8334: URL: https://github.com/apache/paimon/pull/8334#discussion_r3888774681
########## paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionReadPlanner.java: ########## @@ -0,0 +1,329 @@ +/* + * 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.paimon.operation; + +import org.apache.paimon.reader.DataEvolutionRow; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * Pure (no-IO) planner for sub-field-level data evolution reads. Given the requested read row type + * and, for each column-group file ("bunch"), the row type it physically provides (its written + * columns, already wrapped with row-tracking fields), it decides for every read field whether it is + * taken whole from a single file or composed sub-field by sub-field across several files (latest + * file wins per leaf), and produces the offset maps and the per-field {@link + * DataEvolutionRow.NestedField} assembly plans. + * + * <p>Separating this from {@link DataEvolutionSplitRead} keeps the reader-building (IO) thin and + * lets the layout logic be unit-tested directly. Only one level of nested composition is supported; + * deeper or cross-file splits of a sub-struct throw {@link UnsupportedOperationException}. + */ +class DataEvolutionReadPlanner { + + private final RowType readRowType; + // for each bunch, the (row-tracked) row type it physically provides + private final List<RowType> bunchAvailTypes; + + DataEvolutionReadPlanner(RowType readRowType, List<RowType> bunchAvailTypes) { + this.readRowType = readRowType; + this.bunchAvailTypes = bunchAvailTypes; + } + + DataEvolutionReadPlan plan() { + List<DataField> allReadFields = readRowType.getFields(); + int numFields = allReadFields.size(); + int numBunches = bunchAvailTypes.size(); + + // gather, per bunch, the set of leaf field ids it physically provides + List<Set<Integer>> bunchLeaves = new ArrayList<>(); + for (int i = 0; i < numBunches; i++) { + Set<Integer> leaves = new HashSet<>(); + collectLeafIds(bunchAvailTypes.get(i).getFields(), leaves); + bunchLeaves.add(leaves); + } + + // decide, per read field, whether it is taken whole from one file or composed from several + // files at sub-field granularity. Files are already sorted latest-first, so the first bunch + // providing a leaf wins (latest-wins semantics, now at sub-field level). + // selection per bunch: topFieldId -> null (whole) or set of selected sub-field ids + List<Map<Integer, Set<Integer>>> bunchSelection = new ArrayList<>(); + for (int i = 0; i < numBunches; i++) { + bunchSelection.add(new LinkedHashMap<>()); + } + + int[] rowOffsets = new int[numFields]; + int[] fieldOffsets = new int[numFields]; + Arrays.fill(rowOffsets, -1); + Arrays.fill(fieldOffsets, -1); + DataEvolutionRow.NestedField[] nested = new DataEvolutionRow.NestedField[numFields]; + boolean[] composite = new boolean[numFields]; + int[] wholeBunch = new int[numFields]; + Arrays.fill(wholeBunch, -1); + + for (int j = 0; j < numFields; j++) { + DataField rf = allReadFields.get(j); + List<Integer> leaves = leafIdsOf(rf); + Map<Integer, Integer> leafProvider = new HashMap<>(); + Set<Integer> providers = new HashSet<>(); + for (int leaf : leaves) { + int p = providerOf(leaf, bunchLeaves); + if (p >= 0) { + leafProvider.put(leaf, p); + providers.add(p); + } + } + if (providers.isEmpty()) { + // no file provides this field; it stays null (nullability checked below) + continue; + } + // Only read a field whole from a single file when that file covers ALL of its leaves. + // If a single file provides only some leaves of a struct (the rest absent everywhere), + // we must prune to the provided sub-fields so the reader is not asked for sub-fields + // the + // file does not physically contain; the missing ones stay null via the composite plan. + boolean allLeavesCovered = leafProvider.size() == leaves.size(); + if (providers.size() == 1 && allLeavesCovered) { + int b = providers.iterator().next(); + bunchSelection.get(b).put(rf.id(), null); + wholeBunch[j] = b; + } else { + checkArgument( + rf.type() instanceof RowType, + "Field %s is split across files but is not a struct.", + rf.name()); + composite[j] = true; + for (DataField sub : ((RowType) rf.type()).getFields()) { + List<Integer> subLeaves = leafIdsOf(sub); + Set<Integer> subProviders = new HashSet<>(); + int coveredSubLeaves = 0; + for (int leaf : subLeaves) { + int p = leafProvider.getOrDefault(leaf, -1); + if (p >= 0) { + subProviders.add(p); + coveredSubLeaves++; + } + } + if (subProviders.size() > 1) { + throw new UnsupportedOperationException( + "Sub-field-level data evolution does not yet support splitting a " + + "nested sub-field (" + + rf.name() + + "." + + sub.name() + + ") across multiple files."); + } + if (subProviders.size() == 1) { + if (sub.type() instanceof RowType && coveredSubLeaves < subLeaves.size()) { + // the single provider holds only part of this nested sub-struct; + // reading + // it whole would request leaves it lacks, and one-level composition + // cannot prune deeper than this level yet + throw new UnsupportedOperationException( Review Comment: [P1] Preserve deep nested schema evolution. A direct child ROW can be only partially covered because the source file predates a newly added nullable nested field, not because that child was written through an unsupported deep partial path. For example, after adding payload.inner.y to files that contain only payload.inner.x, any overlapping top-level partial update sends the group through this planner and this branch makes full reads and compaction fail. When all physically present leaves of the child come from one bunch and the remaining leaves are absent everywhere, please read the child from that provider and let the existing schema-evolution mapping null-fill the missing leaves. Reserve this exception for an actual cross-provider deep split, and add a regression covering deep ADD COLUMN plus an unrelated overlapping partial update. -- 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]
