This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3bb0104f51 [core] Cache projected schemas for data evolution stats
(#9046)
3bb0104f51 is described below
commit 3bb0104f517a1f1e21372b824bd8176eef464634
Author: YeJunHao <[email protected]>
AuthorDate: Thu Aug 6 13:24:49 2026 +0800
[core] Cache projected schemas for data evolution stats (#9046)
---
.../operation/DataEvolutionFileStoreScan.java | 73 +++++-----
.../paimon/operation/EvolutionStatsCache.java | 161 +++++++++++++++++++++
.../operation/DataEvolutionFileStoreScanTest.java | 106 +++++++++++++-
3 files changed, 295 insertions(+), 45 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
index a19f02320c..5ae7eb07c4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
@@ -76,6 +76,7 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
// per-file column pruning in postFilterManifestEntries.
private final ConcurrentMap<Pair<Long, List<String>>, Set<Integer>>
fileFieldIdsCache =
new ConcurrentHashMap<>();
+ private final EvolutionStatsCache evolutionStatsCache = new
EvolutionStatsCache();
public DataEvolutionFileStoreScan(
ManifestsReader manifestsReader,
@@ -206,7 +207,8 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
}
private boolean filterByStats(List<ManifestEntry> entries) {
- EvolutionStats stats = evolutionStats(schema, this::scanTableSchema,
entries);
+ EvolutionStats stats =
+ evolutionStats(schema, this::scanTableSchema, entries,
evolutionStatsCache);
return inputFilter.test(
stats.rowCount(), stats.minValues(), stats.maxValues(),
stats.nullCounts());
}
@@ -258,19 +260,23 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
pair -> fileFieldIds(this::scanTableSchema, entry.file()));
}
- /** TODO: Optimize implementation of this method. */
@VisibleForTesting
static EvolutionStats evolutionStats(
TableSchema schema,
Function<Long, TableSchema> scanTableSchema,
- List<ManifestEntry> metas) {
+ List<ManifestEntry> metas,
+ EvolutionStatsCache evolutionStatsCache) {
Set<Integer> excludedFileFieldIds =
metas.stream()
.filter(
entry ->
isBlobFile(entry.file().fileName())
||
isVectorStoreFile(entry.file().fileName()))
- .flatMap(entry -> fileFieldIds(scanTableSchema,
entry.file()).stream())
+ .flatMap(
+ entry ->
+
evolutionStatsCache.get(scanTableSchema, entry.file())
+
.dataFileSchema().fields().stream()
+ .map(DataField::id))
.collect(Collectors.toSet());
// exclude blob and vector-store files, useless for predicate eval
metas =
@@ -283,6 +289,8 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
metas.sort(Comparator.comparingLong(maxSeqFunc).reversed());
int[] allFields =
schema.fields().stream().mapToInt(DataField::id).toArray();
+ DataType[] targetTypes =
+
schema.fields().stream().map(DataField::type).toArray(DataType[]::new);
int fieldsCount = schema.fields().size();
int[] rowOffsets = new int[fieldsCount];
int[] fieldOffsets = new int[fieldsCount];
@@ -301,49 +309,38 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
nullCounts[i] = stats.nullCounts();
}
+ int unresolvedFields = fieldsCount;
for (int i = 0; i < metas.size(); i++) {
DataFileMeta fileMeta = metas.get(i).file();
+ EvolutionStatsCache.ProjectedFileSchema projectedFileSchema =
+ evolutionStatsCache.get(scanTableSchema, fileMeta);
- TableSchema dataFileSchema =
-
scanTableSchema.apply(fileMeta.schemaId()).project(fileMeta.writeCols());
-
- TableSchema dataFileSchemaWithStats =
dataFileSchema.project(fileMeta.valueStatsCols());
-
- int[] fieldIds =
- dataFileSchema.logicalRowType().getFields().stream()
- .mapToInt(DataField::id)
- .toArray();
-
- int[] fieldIdsWithStats =
-
dataFileSchemaWithStats.logicalRowType().getFields().stream()
- .mapToInt(DataField::id)
- .toArray();
-
- loop1:
for (int j = 0; j < fieldsCount; j++) {
if (rowOffsets[j] != -1) {
continue;
}
int targetFieldId = allFields[j];
- DataType targetType = schema.fields().get(j).type();
- for (int fieldId : fieldIds) {
- if (targetFieldId == fieldId) {
- for (int k = 0; k < fieldIdsWithStats.length; k++) {
- if (fieldId == fieldIdsWithStats[k]) {
- DataType fileType =
dataFileSchemaWithStats.fields().get(k).type();
- if (!fileType.equalsIgnoreFieldId(targetType))
{
- typeMismatchedFieldIds.add(targetFieldId);
- continue loop1;
- }
- rowOffsets[j] = i;
- fieldOffsets[j] = k;
- continue loop1;
- }
- }
- rowOffsets[j] = -2;
- continue loop1;
- }
+ EvolutionStatsCache.FileFieldStats fileFieldStats =
+ projectedFileSchema.fieldStats(targetFieldId);
+ if (fileFieldStats == null) {
+ continue;
}
+ if (!fileFieldStats.hasStats()) {
+ rowOffsets[j] = -2;
+ unresolvedFields--;
+ continue;
+ }
+ DataType fileType = fileFieldStats.type();
+ if (!fileType.equalsIgnoreFieldId(targetTypes[j])) {
+ typeMismatchedFieldIds.add(targetFieldId);
+ continue;
+ }
+ rowOffsets[j] = i;
+ fieldOffsets[j] = fileFieldStats.index();
+ unresolvedFields--;
+ }
+ if (unresolvedFields == 0) {
+ break;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/EvolutionStatsCache.java
b/paimon-core/src/main/java/org/apache/paimon/operation/EvolutionStatsCache.java
new file mode 100644
index 0000000000..bf1a2a384e
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/EvolutionStatsCache.java
@@ -0,0 +1,161 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Scan-local cache for data evolution stats projections.
+ *
+ * <p>This class is not thread-safe. It is only accessed from the
single-threaded post-filter phase
+ * of {@link AbstractFileStoreScan#plan()}.
+ */
+class EvolutionStatsCache {
+
+ private final Map<CacheKey, ProjectedFileSchema> cache = new HashMap<>();
+
+ ProjectedFileSchema get(Function<Long, TableSchema> scanTableSchema,
DataFileMeta fileMeta) {
+ CacheKey key =
+ new CacheKey(fileMeta.schemaId(), fileMeta.writeCols(),
fileMeta.valueStatsCols());
+ return cache.computeIfAbsent(key, ignored ->
projectFileSchema(scanTableSchema, key));
+ }
+
+ @VisibleForTesting
+ int size() {
+ return cache.size();
+ }
+
+ private static ProjectedFileSchema projectFileSchema(
+ Function<Long, TableSchema> scanTableSchema, CacheKey key) {
+ TableSchema dataFileSchema =
scanTableSchema.apply(key.schemaId).project(key.writeColumns);
+ TableSchema dataFileSchemaWithStats =
dataFileSchema.project(key.valueStatsColumns);
+ List<DataField> fields = dataFileSchema.fields();
+ Map<Integer, FileFieldStats> fieldStats = new HashMap<>(fields.size()
* 2);
+ for (DataField field : fields) {
+ fieldStats.put(field.id(), FileFieldStats.withoutStats());
+ }
+ List<DataField> statsFields = dataFileSchemaWithStats.fields();
+ for (int i = 0; i < statsFields.size(); i++) {
+ DataField statsField = statsFields.get(i);
+ fieldStats.put(statsField.id(), FileFieldStats.withStats(i,
statsField.type()));
+ }
+ return new ProjectedFileSchema(dataFileSchema, fieldStats);
+ }
+
+ static class ProjectedFileSchema {
+
+ private final TableSchema dataFileSchema;
+ private final Map<Integer, FileFieldStats> fieldStats;
+
+ private ProjectedFileSchema(
+ TableSchema dataFileSchema, Map<Integer, FileFieldStats>
fieldStats) {
+ this.dataFileSchema = dataFileSchema;
+ this.fieldStats = fieldStats;
+ }
+
+ TableSchema dataFileSchema() {
+ return dataFileSchema;
+ }
+
+ @Nullable
+ FileFieldStats fieldStats(int fieldId) {
+ return fieldStats.get(fieldId);
+ }
+ }
+
+ static class FileFieldStats {
+
+ private static final FileFieldStats WITHOUT_STATS = new
FileFieldStats(0, null);
+
+ private final int index;
+ @Nullable private final DataType type;
+
+ private FileFieldStats(int index, @Nullable DataType type) {
+ this.index = index;
+ this.type = type;
+ }
+
+ static FileFieldStats withoutStats() {
+ return WITHOUT_STATS;
+ }
+
+ static FileFieldStats withStats(int index, DataType type) {
+ return new FileFieldStats(index, type);
+ }
+
+ boolean hasStats() {
+ return type != null;
+ }
+
+ int index() {
+ checkNotNull(type, "Stats index is unavailable for a field without
stats.");
+ return index;
+ }
+
+ DataType type() {
+ return checkNotNull(type, "Stats type is unavailable for a field
without stats.");
+ }
+ }
+
+ private static class CacheKey {
+
+ private final long schemaId;
+ private final List<String> writeColumns;
+ private final List<String> valueStatsColumns;
+
+ private CacheKey(long schemaId, List<String> writeColumns,
List<String> valueStatsColumns) {
+ this.schemaId = schemaId;
+ this.writeColumns = writeColumns;
+ this.valueStatsColumns = valueStatsColumns;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CacheKey cacheKey = (CacheKey) o;
+ return schemaId == cacheKey.schemaId
+ && Objects.equals(writeColumns, cacheKey.writeColumns)
+ && Objects.equals(valueStatsColumns,
cacheKey.valueStatsColumns);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(schemaId, writeColumns, valueStatsColumns);
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
index 21fbcf629a..647e6a2c8f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
@@ -49,6 +49,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -83,7 +84,10 @@ public class DataEvolutionFileStoreScanTest {
EvolutionStats result =
DataEvolutionFileStoreScan.evolutionStats(
- tableSchema, scanTableSchema,
Collections.singletonList(entry));
+ tableSchema,
+ scanTableSchema,
+ Collections.singletonList(entry),
+ new EvolutionStatsCache());
assertThat(result).isNotNull();
assertThat(result.minValues()).isInstanceOf(DataEvolutionRow.class);
@@ -110,6 +114,87 @@ public class DataEvolutionFileStoreScanTest {
assertThat(maxRow.getFieldCount()).isEqualTo(2);
}
+ @Test
+ public void testEvolutionStatsReusesProjectedSchema() {
+ Schema schema = createSchema("f0", "f1");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+
+ AtomicInteger schemaLoads = new AtomicInteger();
+ Function<Long, TableSchema> countingScanTableSchema =
+ schemaId -> {
+ schemaLoads.incrementAndGet();
+ return schemas.get(schemaId);
+ };
+ EvolutionStatsCache cache = new EvolutionStatsCache();
+ ManifestEntry entry =
+ createManifestEntry(
+ 0L,
+ createSimpleStats(
+ GenericRow.of(1, BinaryString.fromString("a")),
+ GenericRow.of(5, BinaryString.fromString("z")),
+ createBinaryArray(new int[] {0, 1}),
+ new int[] {0, 1}));
+
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema, countingScanTableSchema,
Collections.singletonList(entry), cache);
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema, countingScanTableSchema,
Collections.singletonList(entry), cache);
+
+ assertThat(schemaLoads).hasValue(1);
+ assertThat(cache.size()).isEqualTo(1);
+ }
+
+ @Test
+ public void testEvolutionStatsCacheSeparatesStatsProjections() {
+ Schema schema = createSchema("f0", "f1");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+ EvolutionStatsCache cache = new EvolutionStatsCache();
+
+ ManifestEntry f0StatsEntry =
+ createManifestEntryWithDifferentCols(
+ 0L,
+ new String[] {"f0", "f1"},
+ new String[] {"f0"},
+ createSimpleStats(
+ GenericRow.of(1),
+ GenericRow.of(5),
+ createBinaryArray(new int[] {0}),
+ new int[] {0}));
+ ManifestEntry f1StatsEntry =
+ createManifestEntryWithDifferentCols(
+ 0L,
+ new String[] {"f0", "f1"},
+ new String[] {"f1"},
+ createSimpleStats(
+ GenericRow.of(BinaryString.fromString("a")),
+ GenericRow.of(BinaryString.fromString("z")),
+ createBinaryArray(new int[] {0}),
+ new int[] {1}));
+
+ EvolutionStats f0Stats =
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema,
+ scanTableSchema,
+ Collections.singletonList(f0StatsEntry),
+ cache);
+ EvolutionStats f1Stats =
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema,
+ scanTableSchema,
+ Collections.singletonList(f1StatsEntry),
+ cache);
+
+ DataEvolutionRow f0Min = (DataEvolutionRow) f0Stats.minValues();
+ DataEvolutionRow f1Min = (DataEvolutionRow) f1Stats.minValues();
+ assertThat(f0Min.getInt(0)).isEqualTo(1);
+ assertThat(f0Min.isNullAt(1)).isTrue();
+ assertThat(f1Min.isNullAt(0)).isTrue();
+ assertThat(f1Min.getString(1).toString()).isEqualTo("a");
+ assertThat(cache.size()).isEqualTo(2);
+ }
+
@Test
public void testEvolutionStatsMultipleFiles() {
Schema schema = createSchema("f0", "f1", "f2");
@@ -138,7 +223,8 @@ public class DataEvolutionFileStoreScanTest {
List<ManifestEntry> entries = Arrays.asList(entry2, entry1);
EvolutionStats result =
- DataEvolutionFileStoreScan.evolutionStats(tableSchema,
scanTableSchema, entries);
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema, scanTableSchema, entries, new
EvolutionStatsCache());
assertThat(result).isNotNull();
DataEvolutionRow minRow = (DataEvolutionRow) result.minValues();
@@ -188,7 +274,7 @@ public class DataEvolutionFileStoreScanTest {
EvolutionStats result =
DataEvolutionFileStoreScan.evolutionStats(
- evolvedTableSchema, scanTableSchema, entries);
+ evolvedTableSchema, scanTableSchema, entries, new
EvolutionStatsCache());
assertThat(result).isNotNull();
DataEvolutionRow minRow = (DataEvolutionRow) result.minValues();
@@ -241,7 +327,8 @@ public class DataEvolutionFileStoreScanTest {
List<ManifestEntry> entries = Arrays.asList(entry1, entry2);
EvolutionStats result =
- DataEvolutionFileStoreScan.evolutionStats(tableSchema,
scanTableSchema, entries);
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema, scanTableSchema, entries, new
EvolutionStatsCache());
assertThat(result).isNotNull();
DataEvolutionRow minRow = (DataEvolutionRow) result.minValues();
@@ -303,7 +390,8 @@ public class DataEvolutionFileStoreScanTest {
DataEvolutionFileStoreScan.evolutionStats(
evolvedTableSchema,
scanTableSchema,
- Arrays.asList(oldTypeEntry, newTypeEntry));
+ Arrays.asList(oldTypeEntry, newTypeEntry),
+ new EvolutionStatsCache());
DataEvolutionRow minRow = (DataEvolutionRow) result.minValues();
DataEvolutionRow maxRow = (DataEvolutionRow) result.maxValues();
@@ -338,7 +426,8 @@ public class DataEvolutionFileStoreScanTest {
DataEvolutionFileStoreScan.evolutionStats(
evolvedTableSchema,
scanTableSchema,
- Collections.singletonList(preAlterFile));
+ Collections.singletonList(preAlterFile),
+ new EvolutionStatsCache());
Predicate onChangedColumn =
new
PredicateBuilder(evolvedTableSchema.logicalRowType()).equal(0, 50L);
@@ -387,7 +476,10 @@ public class DataEvolutionFileStoreScanTest {
EvolutionStats result =
DataEvolutionFileStoreScan.evolutionStats(
- tableSchema, scanTableSchema, Arrays.asList(dataEntry,
vectorEntry));
+ tableSchema,
+ scanTableSchema,
+ Arrays.asList(dataEntry, vectorEntry),
+ new EvolutionStatsCache());
DataEvolutionArray nullCounts = (DataEvolutionArray)
result.nullCounts();
assertThat(nullCounts.isNullAt(2)).isTrue();
