You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by ht...@apache.org on 2026/10/05 09:20:48 UTC
(asterixdb) branch master updated: [ASTERIXDB-3935][EXT]: Guard Iceberg string ranges
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new 8098f7d354 [ASTERIXDB-3935][EXT]: Guard Iceberg string ranges
8098f7d354 is described below
commit 8098f7d354228d51c422517425a6d5c44749d5db
Author: Hussain Towaileb <Hu...@Couchbase.com>
AuthorDate: Sat Oct 3 14:05:16 2026 +0300
[ASTERIXDB-3935][EXT]: Guard Iceberg string ranges
- user model changes: no
- storage format changes: no
- interface changes: no
Details: The engine orders strings by UTF-16 code unit, while Iceberg
writes and compares string bounds by code point, so a pushed string
predicate could skip Iceberg files holding matching rows. Pushdown now
declines a string range whose literal holds a character at or above
U+D800, the rule the Parquet row-group filter already applies, shared
between the two as StringOrder. The variant bounds evaluator compares
string bounds by code point and applies the same range rule.
Added tests checking pushed string predicates and variant string bounds
over real Iceberg tables against the engine's comparator, and tripwires
for the Iceberg variant bounds defects still present.
Ext-ref: MB-74379
Generated-by: Claude Opus 5.5 <no...@anthropic.com>
Change-Id: Ib53c1d5785584787df7e4ca3b48386b1b46ebca6
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21987
Integration-Tests: Hussain Towaileb <hu...@gmail.com>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Hussain Towaileb <hu...@gmail.com>
---
.../input/filter/ParquetFilterConverter.java | 22 +-
.../apache/asterix/external/util/StringOrder.java | 45 ++++
.../util/iceberg/VariantBoundsEvaluator.java | 15 +-
.../asterix/external/util/StringOrderTest.java | 137 ++++++++++++
.../util/iceberg/ReadingShreddedPushdownTest.java | 54 +++++
.../external/util/iceberg/StringOrderCorpus.java | 205 ++++++++++++++++++
.../util/iceberg/VariantBoundsStringOrderTest.java | 201 ++++++++++++++++++
asterixdb/asterix-metadata/pom.xml | 7 +
.../utils/filter/IcebergTableFilterBuilder.java | 11 +
.../filter/IcebergStringPredicatePushdownTest.java | 232 +++++++++++++++++++++
10 files changed, 908 insertions(+), 21 deletions(-)
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/filter/ParquetFilterConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/filter/ParquetFilterConverter.java
index 2aa0413726..27ded3521f 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/filter/ParquetFilterConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/filter/ParquetFilterConverter.java
@@ -34,6 +34,7 @@ import org.apache.asterix.external.input.filter.ParquetFilterExpression.Operator
import org.apache.asterix.external.util.ExternalDataConstants.ParquetOptions;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.external.util.MillisecondChronon;
+import org.apache.asterix.external.util.StringOrder;
import org.apache.asterix.external.util.TimestampZoneProjector;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.hadoop.conf.Configuration;
@@ -281,7 +282,7 @@ public class ParquetFilterConverter {
return drop(comparison, "string column compared against " + tag);
}
String literal = (String) value;
- if (operator != Operator.EQ && ordersDifferentlyFromParquet(literal)) {
+ if (operator != Operator.EQ && StringOrder.rangeOrdersDifferently(literal)) {
return drop(comparison, "string range literal holds a character the engine orders differently");
}
return compare(FilterApi.binaryColumn(path), Binary.fromString(literal), operator);
@@ -348,25 +349,6 @@ public class ParquetFilterConverter {
return compare(column, literal, operator);
}
- /**
- * Whether a string range literal could be ordered differently by the engine and by parquet-mr.
- * <p>
- * The engine orders strings by UTF-16 code unit, parquet-mr by unsigned UTF-8 byte, which is code-point order.
- * The two disagree only where a character in U+E000..U+FFFF meets a supplementary character, whose UTF-16 form
- * begins with a surrogate below U+E000. A literal holding no character at or above U+D800 is on the same side
- * of every stored value in both orders -- at the first position they differ its character is below every
- * surrogate -- so only a literal that does hold one needs declining. Equality is unaffected: a row group
- * holding a value always spans it in parquet's own order.
- */
- private static boolean ordersDifferentlyFromParquet(String literal) {
- for (int i = 0; i < literal.length(); i++) {
- if (literal.charAt(i) >= '\uD800') {
- return true;
- }
- }
- return false;
- }
-
private static FilterPredicate intComparison(Comparison comparison, String path, Long value, Operator operator) {
if (value < Integer.MIN_VALUE || value > Integer.MAX_VALUE) {
// the literal lies outside the column's range; clamping it would change which rows match
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/StringOrder.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/StringOrder.java
new file mode 100644
index 0000000000..52af2999ba
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/StringOrder.java
@@ -0,0 +1,45 @@
+/*
+ * 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.asterix.external.util;
+
+/**
+ * Where the engine's string order and code-point order can disagree, for pushing string predicates into readers that
+ * filter by code point: parquet-mr row-group statistics (unsigned UTF-8 bytes) and Iceberg file and row-group bounds.
+ * <p>
+ * The engine orders strings by UTF-16 code unit. The two orders disagree only where a character in U+E000..U+FFFF
+ * meets a supplementary character, whose UTF-16 form begins with a surrogate below U+E000. A range literal holding no
+ * character at or above U+D800 is on the same side of every stored value in both orders -- at the first position they
+ * differ its character is below every surrogate -- so only a literal that does hold one needs declining. Equality is
+ * unaffected: bounds that hold a value always span it in their own order.
+ */
+public final class StringOrder {
+
+ private StringOrder() {
+ }
+
+ /** Whether a range comparison against {@code literal} could exclude a value the engine matches. */
+ public static boolean rangeOrdersDifferently(CharSequence literal) {
+ for (int i = 0; i < literal.length(); i++) {
+ if (literal.charAt(i) >= '\uD800') {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/iceberg/VariantBoundsEvaluator.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/iceberg/VariantBoundsEvaluator.java
index 28f77f4e09..78a29c4c40 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/iceberg/VariantBoundsEvaluator.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/iceberg/VariantBoundsEvaluator.java
@@ -24,6 +24,7 @@ import java.nio.ByteOrder;
import java.util.Map;
import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.external.util.StringOrder;
import org.apache.hyracks.api.exceptions.IWarningCollector;
import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.util.LogRedactionUtil;
@@ -35,6 +36,7 @@ import org.apache.iceberg.expressions.Not;
import org.apache.iceberg.expressions.Or;
import org.apache.iceberg.expressions.UnboundExtract;
import org.apache.iceberg.expressions.UnboundPredicate;
+import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.variants.PhysicalType;
import org.apache.iceberg.variants.Variant;
@@ -194,6 +196,10 @@ public final class VariantBoundsEvaluator {
return true; // no usable bounds for this sub-field (not shredded here, or stats absent)
}
+ if (literal instanceof CharSequence && isRange(predicate.op())
+ && StringOrder.rangeOrdersDifferently((CharSequence) literal)) {
+ return true; // bounds are in code-point order, the engine compares UTF-16 code units
+ }
switch (predicate.op()) {
case EQ:
// cannot match if literal is strictly outside [lower, upper]
@@ -276,6 +282,11 @@ public final class VariantBoundsEvaluator {
}
}
+ private static boolean isRange(Expression.Operation op) {
+ return op == Expression.Operation.LT || op == Expression.Operation.LT_EQ || op == Expression.Operation.GT
+ || op == Expression.Operation.GT_EQ;
+ }
+
// ---- comparisons: every helper returns false when the pair cannot be compared, so callers keep the file ----
private static boolean isLess(Object a, Object b) {
@@ -335,7 +346,9 @@ public final class VariantBoundsEvaluator {
return Long.compare(na.longValue(), nb.longValue());
}
if (a instanceof CharSequence && b instanceof CharSequence) {
- return a.toString().compareTo(b.toString());
+ // the order the bounds were written in; String.compareTo is UTF-16 order and misplaces supplementary
+ // characters against U+E000..U+FFFF
+ return Comparators.charSequences().compare((CharSequence) a, (CharSequence) b);
}
if (a instanceof Boolean && b instanceof Boolean) {
return Boolean.compare((Boolean) a, (Boolean) b);
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/StringOrderTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/StringOrderTest.java
new file mode 100644
index 0000000000..9999e518c1
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/StringOrderTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.asterix.external.util;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hyracks.util.string.UTF8StringUtil;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class StringOrderTest {
+
+ /** Boundary characters on each side of the surrogate range and of U+E000. */
+ private static final List<String> ALPHABET =
+ List.of("a", "z", "é", "中", "", "", "", "＄", "", cp(0x10000), cp(0x1F600), cp(0x10FFFF));
+
+ private static String cp(int codePoint) {
+ return new String(Character.toChars(codePoint));
+ }
+
+ @Test
+ public void flagsOnlyLiteralsWithACharacterAtOrAboveD800() {
+ Assert.assertFalse(StringOrder.rangeOrdersDifferently(""));
+ Assert.assertFalse(StringOrder.rangeOrdersDifferently("abc"));
+ Assert.assertFalse(StringOrder.rangeOrdersDifferently("café"));
+ Assert.assertFalse(StringOrder.rangeOrdersDifferently("中文"));
+ Assert.assertFalse(StringOrder.rangeOrdersDifferently(""));
+ Assert.assertTrue(StringOrder.rangeOrdersDifferently(""));
+ Assert.assertTrue(StringOrder.rangeOrdersDifferently(""));
+ Assert.assertTrue(StringOrder.rangeOrdersDifferently(cp(0x1F600)));
+ Assert.assertTrue(StringOrder.rangeOrdersDifferently("abc＄"));
+ }
+
+ /**
+ * The claim the guard rests on: for a literal it does not flag, the engine's order and code-point order put every
+ * value on the same side of it. Checked exhaustively for every string of up to two characters over the alphabet,
+ * against the engine's own comparator.
+ */
+ @Test
+ public void unflaggedLiteralsCompareTheSameInBothOrders() {
+ List<String> strings = strings(2);
+ int checked = 0;
+ List<String> disagreements = new ArrayList<>();
+ for (String literal : strings) {
+ if (StringOrder.rangeOrdersDifferently(literal)) {
+ continue;
+ }
+ byte[] literalBytes = UTF8StringUtil.writeStringToBytes(literal);
+ for (String value : strings) {
+ int engine = Integer
+ .signum(UTF8StringUtil.compareTo(UTF8StringUtil.writeStringToBytes(value), 0, literalBytes, 0));
+ int codePoint = Integer.signum(compareByCodePoint(value, literal));
+ checked++;
+ if (engine != codePoint && disagreements.size() < 10) {
+ disagreements.add(describe(value) + " vs " + describe(literal));
+ }
+ }
+ }
+ Assert.assertTrue("orders disagree for unflagged literals: " + disagreements, disagreements.isEmpty());
+ Assert.assertTrue(checked > 0);
+ }
+
+ /** The guard is needed: across the same strings, flagged literals do include real disagreements. */
+ @Test
+ public void flaggedLiteralsIncludeDisagreements() {
+ List<String> strings = strings(2);
+ for (String literal : strings) {
+ if (!StringOrder.rangeOrdersDifferently(literal)) {
+ continue;
+ }
+ byte[] literalBytes = UTF8StringUtil.writeStringToBytes(literal);
+ for (String value : strings) {
+ int engine = Integer
+ .signum(UTF8StringUtil.compareTo(UTF8StringUtil.writeStringToBytes(value), 0, literalBytes, 0));
+ if (engine != Integer.signum(compareByCodePoint(value, literal))) {
+ return;
+ }
+ }
+ }
+ Assert.fail("no flagged literal is ever ordered differently, so the guard would be unnecessary");
+ }
+
+ private static List<String> strings(int maxLength) {
+ List<String> result = new ArrayList<>();
+ result.add("");
+ List<String> previous = List.of("");
+ for (int length = 1; length <= maxLength; length++) {
+ List<String> next = new ArrayList<>();
+ for (String prefix : previous) {
+ for (String c : ALPHABET) {
+ next.add(prefix + c);
+ }
+ }
+ result.addAll(next);
+ previous = next;
+ }
+ return result;
+ }
+
+ private static int compareByCodePoint(String a, String b) {
+ int i = 0;
+ int j = 0;
+ while (i < a.length() && j < b.length()) {
+ int ca = a.codePointAt(i);
+ int cb = b.codePointAt(j);
+ if (ca != cb) {
+ return Integer.compare(ca, cb);
+ }
+ i += Character.charCount(ca);
+ j += Character.charCount(cb);
+ }
+ return Integer.compare(a.length() - i, b.length() - j);
+ }
+
+ private static String describe(String s) {
+ StringBuilder sb = new StringBuilder();
+ s.codePoints().forEach(c -> sb.append(String.format("U+%04X ", c)));
+ return sb.length() == 0 ? "\"\"" : sb.toString().trim();
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/ReadingShreddedPushdownTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/ReadingShreddedPushdownTest.java
index a9d7e57fb8..3c461e87ed 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/ReadingShreddedPushdownTest.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/ReadingShreddedPushdownTest.java
@@ -1315,6 +1315,60 @@ public class ReadingShreddedPushdownTest {
}
}
+ /**
+ * TODO(iceberg-15384): tripwire for the second defect — Iceberg's own evaluator cannot read variant bounds once they
+ * have round-tripped through a manifest, because it does not force little-endian. The same file evaluates fine
+ * straight from the writer's metrics (see {@link #scanWorkaround_fullExpressionStillPrunesViaEvaluator}); only the
+ * manifest copy fails. Starts failing when the fix ships, which is the signal to retire VariantBoundsEvaluator.
+ */
+ @Test
+ public void scanWorkaround_icebergCannotReadVariantBoundsFromManifest() throws Exception {
+ org.apache.iceberg.Schema schema = new org.apache.iceberg.Schema(
+ org.apache.iceberg.types.Types.NestedField.required(1, "id",
+ org.apache.iceberg.types.Types.IntegerType.get()),
+ org.apache.iceberg.types.Types.NestedField.optional(2, COLUMN,
+ org.apache.iceberg.types.Types.VariantType.get()));
+ java.io.File dir = java.nio.file.Files.createTempDirectory("scan-workaround-manifest").toFile();
+ org.apache.iceberg.Table table =
+ new org.apache.iceberg.hadoop.HadoopTables(new org.apache.hadoop.conf.Configuration()).create(schema,
+ org.apache.iceberg.PartitionSpec.unpartitioned(),
+ java.util.Map.of(org.apache.iceberg.TableProperties.FORMAT_VERSION, "3"),
+ new java.io.File(dir, "t").getAbsolutePath());
+ table.newAppend().appendFile(writeShreddedBucketFile(schema, 20, 50)).commit();
+ org.apache.iceberg.expressions.Expression outOfRange = VariantPredicateRewriter.rewriteAssumingNesting(
+ org.apache.iceberg.expressions.Expressions.greaterThan(COLUMN + ".amount", 999999), schema);
+ int evaluated = 0;
+ try (org.apache.iceberg.io.CloseableIterable<org.apache.iceberg.FileScanTask> tasks =
+ table.newScan().includeColumnStats().planFiles()) {
+ for (org.apache.iceberg.FileScanTask task : tasks) {
+ evaluated++;
+ try {
+ new org.apache.iceberg.expressions.InclusiveMetricsEvaluator(schema, outOfRange).eval(task.file());
+ Assert.fail("expected Iceberg to fail reading variant bounds back from the manifest");
+ } catch (IllegalArgumentException expected) {
+ Assert.assertTrue(expected.getMessage(),
+ expected.getMessage().contains("Unsupported byte order: big endian"));
+ }
+ }
+ }
+ Assert.assertEquals(1, evaluated);
+ }
+
+ /**
+ * TODO(iceberg-15384): tripwire for the third defect — Iceberg rejects bracket paths, so a sub-field whose name
+ * contains a dot cannot be expressed and VariantPredicateRewriter declines it. Starts failing when bracket
+ * support ships, which is the signal to switch path building to the bracket form.
+ */
+ @Test
+ public void scanWorkaround_icebergRejectsBracketPath() {
+ try {
+ org.apache.iceberg.expressions.Expressions.extract(COLUMN, "$['a.b']", "int");
+ Assert.fail("expected Iceberg to reject a bracket path");
+ } catch (IllegalArgumentException expected) {
+ Assert.assertTrue(expected.getMessage(), expected.getMessage().contains("contains bracket"));
+ }
+ }
+
@Test
public void scanWorkaround_detectsExtractTerms() {
org.apache.iceberg.Schema schema = icebergSchema();
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/StringOrderCorpus.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/StringOrderCorpus.java
new file mode 100644
index 0000000000..02e2e89eb6
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/StringOrderCorpus.java
@@ -0,0 +1,205 @@
+/*
+ * 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.asterix.external.util.iceberg;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hyracks.util.string.UTF8StringUtil;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * Corpus and oracle for the string-ordering pruning tests: {@link VariantBoundsStringOrderTest} here, and
+ * {@code IcebergStringPredicatePushdownTest} in asterix-metadata, which reaches it through this module's test-jar.
+ * <p>
+ * The query engine orders strings by UTF-16 code unit ({@code UTF8StringPointable.compare}), while Iceberg records
+ * and compares string bounds by code point ({@code Comparators.charSequences}). The two orders agree everywhere
+ * except where a supplementary character (stored as a surrogate pair, U+D800..U+DFFF) meets a BMP character in
+ * U+E000..U+FFFF at the first differing position. The corpus is built around that boundary, plus truncation (bounds
+ * are cut at 16 code points) and the empty string. The oracle is the engine's own comparator, never Java's, so a
+ * test's notion of "this row matches" is exactly what the query would return.
+ */
+public final class StringOrderCorpus {
+
+ /** Operators the Iceberg filter builder pushes for a string literal. */
+ public enum Op {
+ EQ,
+ NEQ,
+ LT,
+ LTEQ,
+ GT,
+ GTEQ,
+ STARTS_WITH
+ }
+
+ public static final List<String> VALUES = List.of(
+ // ASCII and the empty string
+ "", "a", "z",
+ // BMP below the surrogate range: both orders agree here
+ "é", "中", "",
+ // BMP above the surrogate range: sorts AFTER a surrogate pair in UTF-16, BEFORE it by code point
+ "", "�", "",
+ // supplementary characters, i.e. surrogate pairs
+ cp(0x10000), cp(0x1F600), cp(0x10FFFF),
+ // the divergence one position in
+ "a", "a" + cp(0x1F600),
+ // longer than the 16-code-point bound truncation, diverging right at the cut
+ "x".repeat(15) + cp(0x1F600) + "tail", "x".repeat(15) + "" + "tail", "x".repeat(20));
+
+ private StringOrderCorpus() {
+ }
+
+ static String cp(int codePoint) {
+ return new String(Character.toChars(codePoint));
+ }
+
+ /** Every single value, then every unordered pair: files of one and of two rows. */
+ public static List<List<String>> fileLayouts() {
+ List<List<String>> layouts = new ArrayList<>();
+ for (String v : VALUES) {
+ layouts.add(List.of(v));
+ }
+ for (int i = 0; i < VALUES.size(); i++) {
+ for (int j = i + 1; j < VALUES.size(); j++) {
+ layouts.add(List.of(VALUES.get(i), VALUES.get(j)));
+ }
+ }
+ return layouts;
+ }
+
+ /** The engine's comparison: what the query evaluates each row with after pushdown has chosen the files. */
+ public static int engineCompare(String a, String b) {
+ return UTF8StringUtil.compareTo(UTF8StringUtil.writeStringToBytes(a), 0, UTF8StringUtil.writeStringToBytes(b),
+ 0);
+ }
+
+ public static boolean engineMatches(String value, Op op, String literal) {
+ switch (op) {
+ case EQ:
+ return engineCompare(value, literal) == 0;
+ case NEQ:
+ return engineCompare(value, literal) != 0;
+ case LT:
+ return engineCompare(value, literal) < 0;
+ case LTEQ:
+ return engineCompare(value, literal) <= 0;
+ case GT:
+ return engineCompare(value, literal) > 0;
+ case GTEQ:
+ return engineCompare(value, literal) >= 0;
+ case STARTS_WITH:
+ return value.startsWith(literal);
+ default:
+ throw new IllegalArgumentException(op.name());
+ }
+ }
+
+ /** {@code column <op> literal} as a plain Iceberg expression, the input the variant rewrite starts from. */
+ public static Expression pushed(String column, Op op, String literal) {
+ switch (op) {
+ case EQ:
+ return Expressions.equal(column, literal);
+ case NEQ:
+ return Expressions.notEqual(column, literal);
+ case LT:
+ return Expressions.lessThan(column, literal);
+ case LTEQ:
+ return Expressions.lessThanOrEqual(column, literal);
+ case GT:
+ return Expressions.greaterThan(column, literal);
+ case GTEQ:
+ return Expressions.greaterThanOrEqual(column, literal);
+ case STARTS_WITH:
+ return Expressions.startsWith(column, literal);
+ default:
+ throw new IllegalArgumentException(op.name());
+ }
+ }
+
+ /**
+ * Whether exact (untruncated) bounds alone prove no row can match: the literal falls outside [min, max]. For the
+ * range operators that is the same as "no row matches"; for EQ a literal strictly between two values matches
+ * nothing yet cannot be excluded by bounds, so it is not counted as prunable.
+ */
+ public static boolean boundsExclude(List<String> values, Op op, String literal) {
+ String min = values.get(0);
+ String max = values.get(0);
+ for (String v : values) {
+ min = engineCompare(v, min) < 0 ? v : min;
+ max = engineCompare(v, max) > 0 ? v : max;
+ }
+ switch (op) {
+ case EQ:
+ return engineCompare(literal, min) < 0 || engineCompare(literal, max) > 0;
+ case LT:
+ return engineCompare(min, literal) >= 0;
+ case LTEQ:
+ return engineCompare(min, literal) > 0;
+ case GT:
+ return engineCompare(max, literal) <= 0;
+ case GTEQ:
+ return engineCompare(max, literal) < 0;
+ default:
+ return false;
+ }
+ }
+
+ /** True when the two orders provably agree for every comparison among these strings and are exact (untruncated). */
+ public static boolean orderInsensitive(List<String> values, String literal) {
+ for (String s : values) {
+ if (!belowSurrogates(s) || s.codePointCount(0, s.length()) > 16) {
+ return false;
+ }
+ }
+ return belowSurrogates(literal) && literal.codePointCount(0, literal.length()) <= 16;
+ }
+
+ private static boolean belowSurrogates(String s) {
+ return s.chars().allMatch(c -> c < 0xD800);
+ }
+
+ public static String describe(String s) {
+ if (s.isEmpty()) {
+ return "\"\"";
+ }
+ StringBuilder sb = new StringBuilder();
+ int asciiRun = 0;
+ for (int i = 0; i < s.length();) {
+ int c = s.codePointAt(i);
+ if (c >= 0x20 && c < 0x7F) {
+ asciiRun++;
+ } else {
+ flush(sb, s, i, asciiRun);
+ asciiRun = 0;
+ sb.append(String.format("U+%04X ", c));
+ }
+ i += Character.charCount(c);
+ }
+ flush(sb, s, s.length(), asciiRun);
+ return sb.toString().trim();
+ }
+
+ private static void flush(StringBuilder sb, String s, int end, int asciiRun) {
+ if (asciiRun > 0) {
+ String run = s.substring(end - asciiRun, end);
+ sb.append(asciiRun > 4 ? "'" + run.charAt(0) + "'x" + asciiRun + " " : "'" + run + "' ");
+ }
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/VariantBoundsStringOrderTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/VariantBoundsStringOrderTest.java
new file mode 100644
index 0000000000..8d6b554b44
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/iceberg/VariantBoundsStringOrderTest.java
@@ -0,0 +1,201 @@
+/*
+ * 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.asterix.external.util.iceberg;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.common.exceptions.WarningCollector;
+import org.apache.asterix.external.util.iceberg.StringOrderCorpus.Op;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * {@link VariantBoundsEvaluator} on shredded STRING sub-fields, over real Iceberg tables whose bounds round-trip
+ * through Avro manifests, checked against the query engine's own string order (see {@link StringOrderCorpus}).
+ * <p>
+ * Every file holds one or two rows, so bounds span a real range: with a single row per file lower equals upper and
+ * the order bounds were computed in can never matter, which is how the single-row type matrix in
+ * {@link VariantBoundsAllTypesTest} misses this. Two properties are checked for every file x operator x literal:
+ * <ul>
+ * <li><b>No data loss</b>: a file holding a row the engine would return is never skipped.</li>
+ * <li><b>Pruning survives</b>: where the two orders provably agree (no character at or above U+D800, nothing
+ * truncated), a file with no matching row is still skipped, so a fix cannot pass by simply giving up on strings.</li>
+ * </ul>
+ */
+public class VariantBoundsStringOrderTest {
+
+ private static final String COLUMN = "variant_field";
+
+ private static Schema schema() {
+ return new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, COLUMN, Types.VariantType.get()));
+ }
+
+ @Test
+ public void topLevelSubField() throws Exception {
+ assertSoundAndEffective("f", List.of("f"));
+ }
+
+ @Test
+ public void nestedSubField() throws Exception {
+ assertSoundAndEffective("a.b", List.of("a", "b"));
+ }
+
+ private static void assertSoundAndEffective(String subField, List<String> path) throws Exception {
+ List<List<String>> layouts = StringOrderCorpus.fileLayouts();
+ Table table = newTable(subField);
+ AppendFiles append = table.newAppend();
+ for (int i = 0; i < layouts.size(); i++) {
+ append.appendFile(writeFile(table, "L" + i, path, layouts.get(i)));
+ }
+ append.commit();
+ Map<String, ContentFile<?>> files = plannedFiles(table);
+ Assert.assertEquals(layouts.size(), files.size());
+
+ List<String> dataLoss = new ArrayList<>();
+ List<String> notPruned = new ArrayList<>();
+ int checks = 0;
+ for (Op op : EnumSet.complementOf(EnumSet.of(Op.STARTS_WITH))) {
+ for (String literal : StringOrderCorpus.VALUES) {
+ Expression rewritten = VariantPredicateRewriter.rewriteAssumingNesting(
+ StringOrderCorpus.pushed(COLUMN + "." + subField, op, literal), schema());
+ VariantBoundsEvaluator evaluator =
+ new VariantBoundsEvaluator(table.schema(), rewritten, new WarningCollector());
+ for (int i = 0; i < layouts.size(); i++) {
+ List<String> rows = layouts.get(i);
+ boolean kept = evaluator.mightMatch(files.get("L" + i + ".parquet"));
+ boolean anyMatch = rows.stream().anyMatch(v -> StringOrderCorpus.engineMatches(v, op, literal));
+ checks++;
+ if (anyMatch && !kept) {
+ dataLoss.add(describe(op, literal, rows));
+ } else if (kept && StringOrderCorpus.orderInsensitive(rows, literal)
+ && StringOrderCorpus.boundsExclude(rows, op, literal)) {
+ notPruned.add(describe(op, literal, rows));
+ }
+ }
+ }
+ }
+ Assert.assertTrue(report("DATA LOSS — file skipped although a row matches", subField, dataLoss, checks),
+ dataLoss.isEmpty());
+ Assert.assertTrue(report("pruning lost where both orders agree", subField, notPruned, checks),
+ notPruned.isEmpty());
+ }
+
+ private static String describe(Op op, String literal, List<String> rows) {
+ StringBuilder sb =
+ new StringBuilder(op.name()).append(' ').append(StringOrderCorpus.describe(literal)).append(" file=[");
+ for (int i = 0; i < rows.size(); i++) {
+ sb.append(i == 0 ? "" : ", ").append(StringOrderCorpus.describe(rows.get(i)));
+ }
+ return sb.append(']').toString();
+ }
+
+ private static String report(String what, String subField, List<String> failures, int checks) {
+ StringBuilder sb = new StringBuilder(what).append(" on '").append(subField).append("': ")
+ .append(failures.size()).append(" of ").append(checks).append(" checks");
+ failures.stream().limit(25).forEach(f -> sb.append("\n ").append(f));
+ return sb.toString();
+ }
+
+ private static Table newTable(String name) throws Exception {
+ File dir = java.nio.file.Files.createTempDirectory("vbso-" + name).toFile();
+ return new HadoopTables(new org.apache.hadoop.conf.Configuration()).create(schema(),
+ PartitionSpec.unpartitioned(), Map.of(TableProperties.FORMAT_VERSION, "3"),
+ new File(dir, "t").getAbsolutePath());
+ }
+
+ /** The variant {@code {path[0]: {path[1]: ... value}}}, shredded down to the leaf. */
+ private static Variant variantAt(List<String> path, String value) {
+ VariantMetadata meta = Variants.metadata(path);
+ VariantValue leaf = Variants.of(value);
+ for (int i = path.size() - 1; i >= 0; i--) {
+ ShreddedObject obj = Variants.object(meta);
+ obj.put(path.get(i), leaf);
+ leaf = obj;
+ }
+ return Variant.of(meta, leaf);
+ }
+
+ private static DataFile writeFile(Table table, String fileName, List<String> path, List<String> values)
+ throws Exception {
+ java.lang.reflect.Method toParquetSchema = Class.forName("org.apache.iceberg.parquet.ParquetVariantUtil")
+ .getDeclaredMethod("toParquetSchema", VariantValue.class);
+ toParquetSchema.setAccessible(true);
+ org.apache.parquet.schema.Type typed =
+ (org.apache.parquet.schema.Type) toParquetSchema.invoke(null, variantAt(path, values.get(0)).value());
+ String location = table.location() + "/data/" + fileName + ".parquet";
+ OutputFile out = table.io().newOutputFile(location);
+ GenericRecord template = GenericRecord.create(table.schema());
+ try (FileAppender<Record> writer = Parquet.write(out).schema(table.schema())
+ .createWriterFunc(GenericParquetWriter::create).variantShreddingFunc((fid, n) -> typed).build()) {
+ for (int id = 0; id < values.size(); id++) {
+ Record row = template.copy();
+ row.setField("id", id);
+ row.setField(COLUMN, variantAt(path, values.get(id)));
+ writer.add(row);
+ }
+ }
+ Metrics metrics = ParquetUtil.fileMetrics(table.io().newInputFile(location), MetricsConfig.forTable(table));
+ return DataFiles.builder(table.spec()).withPath(location).withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(out.toInputFile().getLength()).withMetrics(metrics).build();
+ }
+
+ private static Map<String, ContentFile<?>> plannedFiles(Table table) throws Exception {
+ Map<String, ContentFile<?>> files = new LinkedHashMap<>();
+ try (CloseableIterable<FileScanTask> tasks = table.newScan().includeColumnStats().planFiles()) {
+ for (FileScanTask task : tasks) {
+ files.put(new File(task.file().location()).getName(), task.file().copy());
+ }
+ }
+ return files;
+ }
+}
diff --git a/asterixdb/asterix-metadata/pom.xml b/asterixdb/asterix-metadata/pom.xml
index 160da0e7bf..52326ba43f 100644
--- a/asterixdb/asterix-metadata/pom.xml
+++ b/asterixdb/asterix-metadata/pom.xml
@@ -88,6 +88,13 @@
<artifactId>asterix-external-data</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.asterix</groupId>
+ <artifactId>asterix-external-data</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hyracks</groupId>
<artifactId>algebricks-common</artifactId>
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/IcebergTableFilterBuilder.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/IcebergTableFilterBuilder.java
index e58719b250..9609f73766 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/IcebergTableFilterBuilder.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/filter/IcebergTableFilterBuilder.java
@@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
import org.apache.asterix.external.input.filter.IcebergTableFilterEvaluatorFactory;
import org.apache.asterix.external.util.MillisecondChronon;
+import org.apache.asterix.external.util.StringOrder;
import org.apache.asterix.om.base.ADate;
import org.apache.asterix.om.base.ADateTime;
import org.apache.asterix.om.base.ADouble;
@@ -318,6 +319,11 @@ public class IcebergTableFilterBuilder extends AbstractFilterBuilder {
tryGetLiteralTypeTag(flipped ? left : right));
}
+ private static boolean isRange(FunctionIdentifier fid) {
+ return fid.equals(AlgebricksBuiltinFunctions.LT) || fid.equals(AlgebricksBuiltinFunctions.LE)
+ || fid.equals(AlgebricksBuiltinFunctions.GT) || fid.equals(AlgebricksBuiltinFunctions.GE);
+ }
+
/** @return the type tag behind a constant literal, or {@code null} if the expression is not one. */
private ATypeTag tryGetLiteralTypeTag(ILogicalExpression expression) {
if (expression.getExpressionTag() != LogicalExpressionTag.CONSTANT) {
@@ -351,6 +357,11 @@ public class IcebergTableFilterBuilder extends AbstractFilterBuilder {
}
return buildTruncatedTemporalComparison(fid, columnName, (Long) value, literalTag == ATypeTag.TIME);
}
+ if (value instanceof String && isRange(fid) && StringOrder.rangeOrdersDifferently((String) value)) {
+ // Iceberg prunes files and row groups by code point; the engine compares by UTF-16 code unit
+ LOGGER.debug("String range literal holds a character the engine orders differently; skipping pushdown");
+ return null;
+ }
if (fid.equals(AlgebricksBuiltinFunctions.EQ)) {
return Expressions.equal(columnName, value);
} else if (fid.equals(AlgebricksBuiltinFunctions.NEQ)) {
diff --git a/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/utils/filter/IcebergStringPredicatePushdownTest.java b/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/utils/filter/IcebergStringPredicatePushdownTest.java
new file mode 100644
index 0000000000..8f51aa0d3d
--- /dev/null
+++ b/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/utils/filter/IcebergStringPredicatePushdownTest.java
@@ -0,0 +1,232 @@
+/*
+ * 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.asterix.metadata.utils.filter;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+import org.apache.asterix.external.util.iceberg.StringOrderCorpus;
+import org.apache.asterix.external.util.iceberg.StringOrderCorpus.Op;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetUtil;
+import org.apache.iceberg.types.Types;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * String predicates pushed into Iceberg for an ordinary (non-variant) column, checked against the query engine's own
+ * string order (see {@link StringOrderCorpus}).
+ * <p>
+ * {@link IcebergTableFilterBuilder} pushes {@code =, !=, <, <=, >, >=} and {@code starts_with} on strings. Iceberg then skips data at
+ * two levels before the engine sees a row, and both compare in code-point order: whole files at scan planning
+ * ({@code InclusiveMetricsEvaluator} over manifest bounds), and row groups inside each file (the metrics and dictionary
+ * row-group filters, applied through {@code Parquet.read().filter(residual)}). The engine then re-evaluates the
+ * predicate on what is left, in UTF-16 order. A row the engine would return but either level skipped is lost.
+ * <p>
+ * The predicate comes from the builder itself (see {@link #pushedByBuilder}), so a builder that declines to
+ * push an unsafe one passes. Both levels are then driven as production drives them: planning through
+ * {@code TableScan.filter(..)}, and the read through the same {@code Parquet.read(..).filter(residual)} call as
+ * {@code IcebergFileRecordReader.openStandardRead}, which is package-private to the reader.
+ */
+public class IcebergStringPredicatePushdownTest {
+
+ private static final List<List<String>> LAYOUTS = StringOrderCorpus.fileLayouts();
+ private static Table table;
+
+ @BeforeClass
+ public static void writeTable() throws Exception {
+ Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()), Types.NestedField.optional(3, "st",
+ Types.StructType.of(Types.NestedField.optional(4, "s", Types.StringType.get()))));
+ File dir = java.nio.file.Files.createTempDirectory("string-pushdown-order").toFile();
+ table = new HadoopTables(new org.apache.hadoop.conf.Configuration()).create(schema,
+ PartitionSpec.unpartitioned(), Map.of(TableProperties.FORMAT_VERSION, "2"),
+ new File(dir, "t").getAbsolutePath());
+ AppendFiles append = table.newAppend();
+ for (int i = 0; i < LAYOUTS.size(); i++) {
+ append.appendFile(writeFile(i, LAYOUTS.get(i)));
+ }
+ append.commit();
+ }
+
+ @Test
+ public void topLevelColumn() throws Exception {
+ assertNoRowLost("s", r -> (String) r.getField("s"));
+ }
+
+ @Test
+ public void columnInsideStruct() throws Exception {
+ assertNoRowLost("st.s", r -> (String) ((Record) r.getField("st")).getField("s"));
+ }
+
+ private static void assertNoRowLost(String column, Function<Record, String> valueOf) throws Exception {
+ List<String> lostAtPlanning = new ArrayList<>();
+ List<String> lostInReader = new ArrayList<>();
+ List<String> notPruned = new ArrayList<>();
+ int checks = 0;
+ for (Op op : Op.values()) {
+ for (String literal : StringOrderCorpus.VALUES) {
+ Expression pushed = pushedByBuilder(column, op, literal);
+ if (pushed == null) {
+ pushed = Expressions.alwaysTrue();
+ }
+ Set<String> planned = new HashSet<>();
+ Set<String> returned = new HashSet<>();
+ try (CloseableIterable<FileScanTask> tasks = table.newScan().filter(pushed).planFiles()) {
+ for (FileScanTask task : tasks) {
+ String file = new File(task.file().location()).getName();
+ planned.add(file);
+ try (CloseableIterable<Record> rows =
+ Parquet.read(table.io().newInputFile(task.file().location())).project(table.schema())
+ .filter(task.residual()).split(task.start(), task.length())
+ .createReaderFunc(fs -> GenericParquetReaders.buildReader(table.schema(), fs))
+ .build()) {
+ for (Record r : rows) {
+ returned.add(file + "#" + r.getField("id"));
+ }
+ }
+ }
+ }
+ for (int i = 0; i < LAYOUTS.size(); i++) {
+ List<String> values = LAYOUTS.get(i);
+ String file = "L" + i + ".parquet";
+ checks++;
+ for (int id = 0; id < values.size(); id++) {
+ if (!StringOrderCorpus.engineMatches(values.get(id), op, literal)) {
+ continue;
+ }
+ if (!planned.contains(file)) {
+ lostAtPlanning.add(describe(column, op, literal, values, id));
+ } else if (!returned.contains(file + "#" + id)) {
+ lostInReader.add(describe(column, op, literal, values, id));
+ }
+ }
+ if (planned.contains(file) && StringOrderCorpus.orderInsensitive(values, literal)
+ && StringOrderCorpus.boundsExclude(values, op, literal)) {
+ notPruned.add(describe(column, op, literal, values, -1));
+ }
+ }
+ }
+ }
+ Assert.assertTrue(report("DATA LOSS at scan planning (file skipped)", lostAtPlanning, checks),
+ lostAtPlanning.isEmpty());
+ Assert.assertTrue(report("DATA LOSS in the reader (row group skipped)", lostInReader, checks),
+ lostInReader.isEmpty());
+ Assert.assertTrue(report("pruning lost where both orders agree", notPruned, checks), notPruned.isEmpty());
+ }
+
+ /**
+ * The predicate the filter builder pushes for {@code column <op> literal}, or {@code null} when it declines to push
+ * one. Comparisons come from the builder itself; {@code starts_with} is built inline, as the builder's own
+ * handler needs a logical plan, and is the builder's {@code Expressions.startsWith(column, prefix)} verbatim.
+ */
+ private static Expression pushedByBuilder(String column, Op op, String literal) {
+ switch (op) {
+ case EQ:
+ return IcebergTableFilterBuilder.buildComparisonExpression(AlgebricksBuiltinFunctions.EQ, column,
+ literal, ATypeTag.STRING);
+ case NEQ:
+ return IcebergTableFilterBuilder.buildComparisonExpression(AlgebricksBuiltinFunctions.NEQ, column,
+ literal, ATypeTag.STRING);
+ case LT:
+ return IcebergTableFilterBuilder.buildComparisonExpression(AlgebricksBuiltinFunctions.LT, column,
+ literal, ATypeTag.STRING);
+ case LTEQ:
+ return IcebergTableFilterBuilder.buildComparisonExpression(AlgebricksBuiltinFunctions.LE, column,
+ literal, ATypeTag.STRING);
+ case GT:
+ return IcebergTableFilterBuilder.buildComparisonExpression(AlgebricksBuiltinFunctions.GT, column,
+ literal, ATypeTag.STRING);
+ case GTEQ:
+ return IcebergTableFilterBuilder.buildComparisonExpression(AlgebricksBuiltinFunctions.GE, column,
+ literal, ATypeTag.STRING);
+ case STARTS_WITH:
+ return Expressions.startsWith(column, literal);
+ default:
+ throw new IllegalArgumentException(op.name());
+ }
+ }
+
+ private static String describe(String column, Op op, String literal, List<String> values, int row) {
+ StringBuilder sb = new StringBuilder(column).append(' ').append(op.name()).append(' ')
+ .append(StringOrderCorpus.describe(literal)).append(" file=[");
+ for (int i = 0; i < values.size(); i++) {
+ sb.append(i == 0 ? "" : ", ").append(i == row ? "*" : "").append(StringOrderCorpus.describe(values.get(i)));
+ }
+ return sb.append(']').toString();
+ }
+
+ private static String report(String what, List<String> failures, int checks) {
+ StringBuilder sb = new StringBuilder(what).append(": ").append(failures.size()).append(" (over ").append(checks)
+ .append(" file checks; * marks the lost row)");
+ failures.stream().limit(25).forEach(f -> sb.append("\n ").append(f));
+ return sb.toString();
+ }
+
+ private static DataFile writeFile(int index, List<String> values) throws Exception {
+ String location = table.location() + "/data/L" + index + ".parquet";
+ OutputFile out = table.io().newOutputFile(location);
+ GenericRecord template = GenericRecord.create(table.schema());
+ GenericRecord struct = GenericRecord.create(table.schema().findType("st").asStructType());
+ try (FileAppender<Record> writer =
+ Parquet.write(out).schema(table.schema()).createWriterFunc(GenericParquetWriter::create).build()) {
+ for (int id = 0; id < values.size(); id++) {
+ Record row = template.copy();
+ row.setField("id", id);
+ row.setField("s", values.get(id));
+ Record st = struct.copy();
+ st.setField("s", values.get(id));
+ row.setField("st", st);
+ writer.add(row);
+ }
+ }
+ Metrics metrics = ParquetUtil.fileMetrics(table.io().newInputFile(location), MetricsConfig.forTable(table));
+ return DataFiles.builder(table.spec()).withPath(location).withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(out.toInputFile().getLength()).withMetrics(metrics).build();
+ }
+}