HIVE-30176: Iceberg: file-level pruning of shredded VARIANT values

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java

@@ -185,6 +185,14 @@ public static Expression residualForReaderPruning(FileScanTask task, Configurati
     return FilterExpressions.readerResidual(task, conf);
   }
 
+  /**
+   * Returns the Iceberg filter of the pushed-down Hive predicate, including {@code extract(...)} predicates;
+   * intended only for pruning.
+   */
+  public static Expression readerFilter(Configuration conf) {
+    return FilterExpressions.icebergDataFilterFromHiveConf(conf);
+  }
+
   @Override
   public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
     Expression filter = FilterExpressions.planningFilter(job);

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/variant/VariantFilePruner.java

@@ -0,0 +1,149 @@
+/*
+ * 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.iceberg.mr.hive.variant;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.And;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
+import org.apache.iceberg.expressions.Not;
+import org.apache.iceberg.expressions.UnboundExtract;
+import org.apache.iceberg.expressions.UnboundPredicate;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+/**
+ * Drops data files whose variant bounds show that no row can match the {@code extract(...)} predicates of a
+ * scan filter.
+ *
+ * <p>A bridge until apache/iceberg#18518 is in Hive's Iceberg version. Before it, Iceberg planning cannot
+ * describe an {@code extract(...)} term and reads the little-endian variant bounds of a manifest in a ByteBuffer's
+ * default big-endian order, so Hive strips these predicates from the planning filter. Once #18518 is available,
+ * keep {@code extract(...)} in the planning filter and delete this class and its hook in {@code IcebergInputFormat}.
+ *
+ * <p>Variant string bounds written by Iceberg 1.10 and 1.11 can be inverted (apache/iceberg#17397); they are
+ * trusted here, as Iceberg's own planning does.
+ */
+public final class VariantFilePruner {
+
+  private final InclusiveMetricsEvaluator evaluator;
+  private final Set<Integer> variantFieldIds;
+  private final List<String> variantColumns;
+
+  private VariantFilePruner(Schema schema, Expression extractFilter, boolean caseSensitive) {
+    this.evaluator = new InclusiveMetricsEvaluator(schema, extractFilter, caseSensitive);
+    this.variantFieldIds = Binder.boundReferences(schema.asStruct(), extractFilter, caseSensitive);
+    this.variantColumns = variantFieldIds.stream().map(schema::findColumnName).toList();
+  }
+
+  /**
+   * Returns a pruner for the top-level conjuncts of the filter that are an {@code extract(...)} predicate, or the NOT
+   * of one, on a variant column of the scan's schema; null if there are none. A time-travel scan can predate the
+   * variant column, and {@code variant_get} also accepts struct columns.
+   *
+   * <p>Hive's search argument writes {@code >}, {@code >=}, {@code <>} and {@code IS NOT NULL} as the NOT of a
+   * predicate, which the evaluator negates exactly. All other conjuncts, OR, and NOT over anything else are ignored,
+   * which only keeps more files.
+   */
+  public static VariantFilePruner of(Schema scanSchema, Expression filter, boolean caseSensitive) {
+    Predicate<String> isVariantColumn = column -> {
+      Types.NestedField field = caseSensitive ?
+          scanSchema.findField(column) : scanSchema.caseInsensitiveFindField(column);
+      return field != null && field.type().typeId() == Type.TypeID.VARIANT;
+    };
+    Expression extractFilter = extractConjuncts(filter, isVariantColumn);
+    if (extractFilter == Expressions.alwaysTrue()) {
+      return null;
+    }
+    return new VariantFilePruner(scanSchema, extractFilter, caseSensitive);
+  }
+
+  private static Expression extractConjuncts(Expression expr, Predicate<String> isVariantColumn) {
+    if (expr instanceof And and) {
+      return Expressions.and(
+          extractConjuncts(and.left(), isVariantColumn), extractConjuncts(and.right(), isVariantColumn));
+    }
+    if (isExtractPredicate(expr, isVariantColumn) ||
+        expr instanceof Not not && isExtractPredicate(not.child(), isVariantColumn)) {
+      return expr;
+    }
+    return Expressions.alwaysTrue();
+  }
+
+  private static boolean isExtractPredicate(Expression expr, Predicate<String> isVariantColumn) {
+    return expr instanceof UnboundPredicate<?> predicate && predicate.term() instanceof UnboundExtract<?> extract &&
+        isVariantColumn.test(extract.ref().name());
+  }
+
+  /**
+   * The variant columns whose stats {@link #mayMatch(FileScanTask)} needs; planned files carry no stats unless
+   * the scan requests them.
+   */
+  public List<String> variantColumns() {
+    return variantColumns;
+  }
+
+  /**
+   * Returns false if no row of the task's file can match.
+   */
+  public boolean mayMatch(FileScanTask task) {
+    DataFile file = task.file();
+    if (!hasVariantBounds(file.lowerBounds()) && !hasVariantBounds(file.upperBounds())) {
+      return evaluator.eval(file);
+    }
+    DataFile withVariantBounds = DataFiles.builder(task.spec())
+        .copy(file)
+        .withMetrics(new Metrics(file.recordCount(), file.columnSizes(), file.valueCounts(),
+            file.nullValueCounts(), file.nanValueCounts(),
+            littleEndian(file.lowerBounds()), littleEndian(file.upperBounds())))
+        .build();
+    return evaluator.eval(withVariantBounds);
+  }
+
+  private boolean hasVariantBounds(Map<Integer, ByteBuffer> bounds) {
+    return bounds != null && variantFieldIds.stream().anyMatch(bounds::containsKey);
+  }
+
+  /**
+   * Serialized variants are little-endian; a ByteBuffer's byte order defaults to big-endian.
+   */
+  private Map<Integer, ByteBuffer> littleEndian(Map<Integer, ByteBuffer> bounds) {
+    if (bounds == null) {
+      return null;
+    }
+    Map<Integer, ByteBuffer> view = Maps.newHashMap(bounds);
+    variantFieldIds.forEach(id ->
+        view.computeIfPresent(id, (key, buffer) -> buffer.duplicate().order(ByteOrder.LITTLE_ENDIAN)));
+    return view;
+  }
+}

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java

@@ -58,7 +58,9 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.mr.Catalogs;
 import org.apache.iceberg.mr.InputFormatConfig;
+import org.apache.iceberg.mr.hive.HiveIcebergInputFormat;
 import org.apache.iceberg.mr.hive.HiveTableUtil;
+import org.apache.iceberg.mr.hive.variant.VariantFilePruner;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.SerializationUtil;
@@ -192,13 +194,19 @@ private List<InputSplit> planInputSplits(Table table, Configuration conf, Execut
     }
     scan = scan.planWith(workerPool);
 
+    VariantFilePruner variantPruner =
+        VariantFilePruner.of(scan.schema(), HiveIcebergInputFormat.readerFilter(conf), scan.isCaseSensitive());
+    if (variantPruner != null) {
+      scan = scan.includeColumnStats(variantPruner.variantColumns());
+    }
+
     boolean allowDataFilesWithinTableLocationOnly =
         conf.getBoolean(HiveConf.ConfVars.HIVE_ICEBERG_ALLOW_DATAFILES_IN_TABLE_LOCATION_ONLY.varname,
             HiveConf.ConfVars.HIVE_ICEBERG_ALLOW_DATAFILES_IN_TABLE_LOCATION_ONLY.defaultBoolVal);
     Path tableLocation = new Path(conf.get(InputFormatConfig.TABLE_LOCATION));
 
     String[] groupingPartitionColumns = conf.getStrings(InputFormatConfig.GROUPING_PARTITION_COLUMNS);
-    generateInputSplits(scan, table, groupingPartitionColumns, taskGroup -> {
+    generateInputSplits(scan, table, groupingPartitionColumns, variantPruner, taskGroup -> {
       if (applyResidual && model == InputFormatConfig.InMemoryDataModel.HIVE) {
         // TODO: We do not support residual evaluation for HIVE and PIG in memory data model yet
         checkResiduals(taskGroup);
@@ -227,17 +235,18 @@ private static void validateFileLocations(ScanTaskGroup<FileScanTask> split, Pat
   }
 
   private static void generateInputSplits(Scan<?, FileScanTask, CombinedScanTask> scan, Table table,
-      String[] groupingPartitionColumns, Consumer<ScanTaskGroup<FileScanTask>> consumer) {
+      String[] groupingPartitionColumns, VariantFilePruner variantPruner,
+      Consumer<ScanTaskGroup<FileScanTask>> consumer) {
     if (groupingPartitionColumns == null) {
-      try (CloseableIterable<CombinedScanTask> tasksIterable = scan.planTasks()) {
+      try (CloseableIterable<CombinedScanTask> tasksIterable = planTasks(scan, variantPruner)) {
         tasksIterable.forEach(consumer);
       } catch (IOException e) {
         throw new UncheckedIOException(String.format("Failed to close table scan: %s", scan), e);
       }
     } else {
       final StructType groupingKeyType = Partitioning.groupingKeyType(
           table.schema().select(groupingPartitionColumns), table.specs().values());
-      try (CloseableIterable<FileScanTask> taskIterable = scan.planFiles()) {
+      try (CloseableIterable<FileScanTask> taskIterable = planFiles(scan, variantPruner)) {
         final List<FileScanTask> tasks = Lists.newArrayList(taskIterable);
         final List<ScanTaskGroup<FileScanTask>> partitionScanTaskGroups =
             TableScanUtil.planTaskGroups(
@@ -253,6 +262,26 @@ private static void generateInputSplits(Scan<?, FileScanTask, CombinedScanTask>
     }
   }
 
+  /**
+   * Plans the files left after variant pruning into tasks with the same split sizing as {@code scan.planTasks()}.
+   */
+  private static CloseableIterable<CombinedScanTask> planTasks(Scan<?, FileScanTask, CombinedScanTask> scan,
+      VariantFilePruner variantPruner) {
+    if (variantPruner == null) {
+      return scan.planTasks();
+    }
+    CloseableIterable<FileScanTask> splitFiles =
+        TableScanUtil.splitFiles(planFiles(scan, variantPruner), scan.targetSplitSize());
+    return TableScanUtil.planTasks(
+        splitFiles, scan.targetSplitSize(), scan.splitLookback(), scan.splitOpenFileCost());
+  }
+
+  private static CloseableIterable<FileScanTask> planFiles(Scan<?, FileScanTask, CombinedScanTask> scan,
+      VariantFilePruner variantPruner) {
+    CloseableIterable<FileScanTask> files = scan.planFiles();
+    return variantPruner == null ? files : CloseableIterable.filter(files, variantPruner::mayMatch);
+  }
+
   private static void checkResiduals(ScanTaskGroup<FileScanTask> task) {
     task.tasks().forEach(fileScanTask -> {
       Expression residual = fileScanTask.residual();