From 6efe6adb0ecf5c711d4b2fab30a4078bb825f370 Mon Sep 17 00:00:00 2001 From: Raghvendra Singh Date: Wed, 2 Sep 2026 02:04:07 +0530 Subject: [PATCH] [fix](iceberg) Project the row filter before the delete-manifest prune in the cached scan plan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cacheBackedFileScanTasks (the manifest-cache planning path shared by the synchronous, streaming and COUNT(*) plans) prunes DELETE manifests with ManifestEvaluator.forPartitionFilter(filterExpr, spec, caseSensitive) passing the raw ROW filter. forPartitionFilter binds against the partition struct, so this works by accident on identity-only specs (the partition field keeps the source column name) and throws ValidationException ("Cannot find field 'ts' in struct: struct<... ts_month: int>") on any spec with a transform — e.g. (identity(flag), month(ts)) — whenever the filter references the transform's source column. The catch then aborts the WHOLE cached plan into the SDK fallback, logging a WARN with a stack trace per query: [IcebergScanPlanProvider.planFileScanTask()] Iceberg plan with manifest cache failed, falling back to SDK scan: Cannot find field 'ts' in struct: struct<1000: flag: optional boolean, 1001: ts_month: optional int> So on v2 tables with delete files and a time-transform partition spec, every filtered query silently loses the manifest cache (planning latency + repeated catalog/storage manifest reads) and spams the warn log — we measured ~1,100 such stacks per hour on one production FE where most queries filter on the transform's source column. The data-manifest side of the very same method (getMatchingManifest) already projects the filter into partition space before building its evaluator. Do the same for delete manifests: project with Projections.inclusive(spec, caseSensitive) — predicates on non-partition columns project to alwaysTrue(), so pruning semantics are unchanged. The evaluator construction moves into a small package-private helper so the behavior is unit-testable. The new test builds a real InMemoryCatalog v2 table partitioned by (identity(flag), month(ts)) with a position-delete file and pins: the raw row filter still fails to bind (the projection stays load-bearing), the projected evaluator keeps an overlapping month and prunes a far month, the identity leg still prunes, and a residual-only filter keeps the manifest. Signed-off-by: Raghvendra Singh Co-Authored-By: Claude Fable 5 --- .../iceberg/IcebergScanPlanProvider.java | 19 ++- ...anPlanProviderDeleteManifestPruneTest.java | 152 ++++++++++++++++++ 2 files changed, 170 insertions(+), 1 deletion(-) create mode 100644 fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderDeleteManifestPruneTest.java diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java index 4ec7649b514844..a766ab91a58b96 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java @@ -2758,7 +2758,7 @@ private CloseableIterable cacheBackedFileScanTasks(TableScan scan, if (spec == null) { continue; } - if (!ManifestEvaluator.forPartitionFilter(filterExpr, spec, caseSensitive).eval(manifest)) { + if (!deleteManifestEvaluator(spec, filterExpr, caseSensitive).eval(manifest)) { continue; } deleteFiles.addAll(manifestCacheGet(manifest, table, statsQueryId).getDeleteFiles()); @@ -2929,6 +2929,23 @@ private static CloseableIterable getMatchingManifest(List manifest.hasAddedFiles() || manifest.hasExistingFiles()); } + /** + * Partition-prune evaluator for DELETE manifests: the ROW filter must be projected into partition space + * before {@link ManifestEvaluator#forPartitionFilter}, which binds against the partition struct. Passing + * the raw row filter binds fine on identity-only specs (the partition field keeps the source column name) + * but throws {@code ValidationException} ("Cannot find field ...") on any spec with a transform + * (e.g. {@code month(ts)} stores the field as {@code ts_month}) whenever the filter references the + * transform's source column — which aborted the WHOLE cached plan into the SDK fallback for every + * filtered query on such tables (one WARN + stack per query). The data-manifest side + * ({@link #getMatchingManifest}) always projected; this mirrors it. An inclusive projection maps + * predicates on non-partition columns to {@code alwaysTrue()}, so pruning stays correct. + */ + static ManifestEvaluator deleteManifestEvaluator(PartitionSpec spec, Expression rowFilter, + boolean caseSensitive) { + return ManifestEvaluator.forPartitionFilter( + Projections.inclusive(spec, caseSensitive).project(rowFilter), spec, caseSensitive); + } + /** * Port of legacy {@code IcebergUtils.isManifestCacheEnabled}: the manifest-level path is used iff the * manifest cache is wired AND the spec is enabled ({@code enable && ttl-second != 0 && capacity != 0}). diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderDeleteManifestPruneTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderDeleteManifestPruneTest.java new file mode 100644 index 00000000000000..76c4904c9d5745 --- /dev/null +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderDeleteManifestPruneTest.java @@ -0,0 +1,152 @@ +// 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.connector.iceberg; + +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.ManifestEvaluator; +import org.apache.iceberg.inmemory.InMemoryCatalog; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Collections; +import java.util.List; + +/** + * The delete-manifest partition prune inside the manifest-cache planning path must project the ROW filter into + * partition space before {@link ManifestEvaluator#forPartitionFilter} (exactly like the data-manifest side). + * Binding the raw row filter works by accident on identity-only specs (the partition field keeps the source + * column name) but throws on any spec with a transform — e.g. {@code month(ts)} stores the partition field as + * {@code ts_month: int} — whenever the filter references the transform's source column. Before the fix that + * exception aborted the whole cached plan into the SDK fallback for every filtered query on every table with + * delete manifests, logging a WARN + stack per query. + * + *

Uses a real {@link InMemoryCatalog} v2 table (identity + month spec, one data file, one position-delete + * file) so the evaluator runs against a genuine delete manifest with real partition summaries. + */ +public class IcebergScanPlanProviderDeleteManifestPruneTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "flag", Types.BooleanType.get()), + Types.NestedField.required(2, "ts", Types.TimestampType.withoutZone())); + + // month(ts) ordinals are months since 1970-01: 2026-07 -> (2026-1970)*12 + 6 = 678. + private static final String JULY_2026_PATH = "flag=true/ts_month=678"; + + private static long micros(String instant) { + return Instant.parse(instant).toEpochMilli() * 1000L; + } + + /** v2 table partitioned by (identity(flag), month(ts)) with one data + one position-delete file in 2026-07. */ + private static Table tableWithJuly2026Deletes() { + PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("flag").month("ts").build(); + InMemoryCatalog catalog = new InMemoryCatalog(); + catalog.initialize("test", Collections.emptyMap()); + catalog.createNamespace(Namespace.of("db1")); + Table table = catalog.createTable(TableIdentifier.of("db1", "t1"), SCHEMA, spec, + Collections.singletonMap("format-version", "2")); + table.newAppend() + .appendFile(DataFiles.builder(table.spec()) + .withPath("/data/f1.parquet").withFileSizeInBytes(100).withRecordCount(10) + .withPartitionPath(JULY_2026_PATH).withFormat(FileFormat.PARQUET).build()) + .commit(); + table.newRowDelta() + .addDeletes(FileMetadata.deleteFileBuilder(table.spec()) + .ofPositionDeletes() + .withPath("/data/pd1.parquet").withFileSizeInBytes(50).withRecordCount(1) + .withPartitionPath(JULY_2026_PATH).withFormat(FileFormat.PARQUET).build()) + .commit(); + return table; + } + + private static ManifestFile onlyDeleteManifest(Table table) { + List deletes = table.currentSnapshot().deleteManifests(table.io()); + Assertions.assertEquals(1, deletes.size(), "the single row delta produced one delete manifest"); + return deletes.get(0); + } + + @Test + public void rawRowFilterFailsToBindOnTransformSpec() { + // The bug this class guards against: binding the un-projected row filter against the partition struct + // throws, because 'ts' only appears through the month transform ('ts_month'). If this ever STOPS + // throwing, the projection in deleteManifestEvaluator becomes optional — revisit, don't delete it. + Table table = tableWithJuly2026Deletes(); + Expression rowFilter = Expressions.greaterThan("ts", micros("2026-07-15T00:00:00Z")); + Assertions.assertThrows(ValidationException.class, + () -> ManifestEvaluator.forPartitionFilter(rowFilter, table.spec(), true) + .eval(onlyDeleteManifest(table))); + } + + @Test + public void projectedEvaluatorKeepsOverlappingMonthAndPrunesFarMonth() { + Table table = tableWithJuly2026Deletes(); + ManifestFile deleteManifest = onlyDeleteManifest(table); + + // WHY: a filter inside 2026-07 must keep the manifest (its deletes apply to matching data). + // MUTATION: dropping the projection -> ValidationException -> red. + Assertions.assertTrue(IcebergScanPlanProvider.deleteManifestEvaluator( + table.spec(), Expressions.greaterThan("ts", micros("2026-07-15T00:00:00Z")), true) + .eval(deleteManifest)); + + // WHY: a filter entirely after 2026-07 (2027-05 -> ordinal 688) must PRUNE the manifest — the + // projection maps ts > X to ts_month >= 688, and the summaries hold only 678. MUTATION: projecting + // to alwaysTrue() (no prune) -> true -> red. + Assertions.assertFalse(IcebergScanPlanProvider.deleteManifestEvaluator( + table.spec(), Expressions.greaterThan("ts", micros("2027-05-01T00:00:00Z")), true) + .eval(deleteManifest)); + } + + @Test + public void identityPredicateStillPrunes() { + // Unchanged behavior on the identity leg of the same spec: flag=true keeps, flag=false prunes. + Table table = tableWithJuly2026Deletes(); + ManifestFile deleteManifest = onlyDeleteManifest(table); + + Assertions.assertTrue(IcebergScanPlanProvider.deleteManifestEvaluator( + table.spec(), Expressions.equal("flag", true), true).eval(deleteManifest)); + Assertions.assertFalse(IcebergScanPlanProvider.deleteManifestEvaluator( + table.spec(), Expressions.equal("flag", false), true).eval(deleteManifest)); + } + + @Test + public void predicateOnNonPartitionColumnProjectsToKeep() { + // A residual-only filter (no partition column at all) must not prune delete manifests: the inclusive + // projection yields alwaysTrue(). Guard against an over-eager projection that prunes everything. + Schema wider = new Schema( + Types.NestedField.required(1, "flag", Types.BooleanType.get()), + Types.NestedField.required(2, "ts", Types.TimestampType.withoutZone()), + Types.NestedField.optional(3, "note", Types.StringType.get())); + PartitionSpec spec = PartitionSpec.builderFor(wider).identity("flag").month("ts").build(); + Table table = tableWithJuly2026Deletes(); + Assertions.assertTrue(IcebergScanPlanProvider.deleteManifestEvaluator( + spec, Expressions.equal("note", "x"), true) + .eval(onlyDeleteManifest(table))); + } +}