From 079a97ea432c8db5b24d22a96d53f5727b982187 Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Thu, 15 Oct 2026 17:33:45 +0300 Subject: [PATCH] HIVE-30176: Iceberg: file-level pruning of shredded VARIANT values --- .../mr/hive/HiveIcebergInputFormat.java | 8 + .../mr/hive/variant/VariantFilePruner.java | 149 +++++ .../mr/mapreduce/IcebergInputFormat.java | 37 +- .../TestHiveIcebergVariantFilePruning.java | 565 ++++++++++++++++++ .../hive/variant/TestVariantFilePruner.java | 157 +++++ 5 files changed, 912 insertions(+), 4 deletions(-) create mode 100644 iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/variant/VariantFilePruner.java create mode 100644 iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergVariantFilePruning.java create mode 100644 iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/variant/TestVariantFilePruner.java diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java index c7c0557dc6f3..1934805d3b32 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java +++ b/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); diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/variant/VariantFilePruner.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/variant/VariantFilePruner.java new file mode 100644 index 000000000000..e2f488ca92b2 --- /dev/null +++ b/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. + * + *

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}. + * + *

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 variantFieldIds; + private final List 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. + * + *

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 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 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 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 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 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 littleEndian(Map bounds) { + if (bounds == null) { + return null; + } + Map view = Maps.newHashMap(bounds); + variantFieldIds.forEach(id -> + view.computeIfPresent(id, (key, buffer) -> buffer.duplicate().order(ByteOrder.LITTLE_ENDIAN))); + return view; + } +} diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java index 89f888255b89..d3f51655c3bb 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java +++ b/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 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,9 +235,10 @@ private static void validateFileLocations(ScanTaskGroup split, Pat } private static void generateInputSplits(Scan scan, Table table, - String[] groupingPartitionColumns, Consumer> consumer) { + String[] groupingPartitionColumns, VariantFilePruner variantPruner, + Consumer> consumer) { if (groupingPartitionColumns == null) { - try (CloseableIterable tasksIterable = scan.planTasks()) { + try (CloseableIterable tasksIterable = planTasks(scan, variantPruner)) { tasksIterable.forEach(consumer); } catch (IOException e) { throw new UncheckedIOException(String.format("Failed to close table scan: %s", scan), e); @@ -237,7 +246,7 @@ private static void generateInputSplits(Scan } else { final StructType groupingKeyType = Partitioning.groupingKeyType( table.schema().select(groupingPartitionColumns), table.specs().values()); - try (CloseableIterable taskIterable = scan.planFiles()) { + try (CloseableIterable taskIterable = planFiles(scan, variantPruner)) { final List tasks = Lists.newArrayList(taskIterable); final List> partitionScanTaskGroups = TableScanUtil.planTaskGroups( @@ -253,6 +262,26 @@ private static void generateInputSplits(Scan } } + /** + * Plans the files left after variant pruning into tasks with the same split sizing as {@code scan.planTasks()}. + */ + private static CloseableIterable planTasks(Scan scan, + VariantFilePruner variantPruner) { + if (variantPruner == null) { + return scan.planTasks(); + } + CloseableIterable splitFiles = + TableScanUtil.splitFiles(planFiles(scan, variantPruner), scan.targetSplitSize()); + return TableScanUtil.planTasks( + splitFiles, scan.targetSplitSize(), scan.splitLookback(), scan.splitOpenFileCost()); + } + + private static CloseableIterable planFiles(Scan scan, + VariantFilePruner variantPruner) { + CloseableIterable files = scan.planFiles(); + return variantPruner == null ? files : CloseableIterable.filter(files, variantPruner::mayMatch); + } + private static void checkResiduals(ScanTaskGroup task) { task.tasks().forEach(fileScanTask -> { Expression residual = fileScanTask.residual(); diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergVariantFilePruning.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergVariantFilePruning.java new file mode 100644 index 000000000000..b5c6015c053f --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergVariantFilePruning.java @@ -0,0 +1,565 @@ +/* + * 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; + +import java.io.File; +import java.io.IOException; +import java.net.URI; +import java.util.Arrays; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Consumer; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.apache.hadoop.hive.ql.exec.SerializationUtilities; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc; +import org.apache.hadoop.hive.ql.plan.TableDesc; +import org.apache.hadoop.hive.ql.plan.TableScanDesc; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDF; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBetween; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFIn; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPAnd; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqual; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqualOrGreaterThan; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPGreaterThan; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPLessThan; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNot; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNotEqual; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNotNull; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNull; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPOr; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFVariantGet; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; +import org.apache.hadoop.mapred.InputSplit; +import org.apache.hadoop.mapred.JobConf; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.mr.InputFormatConfig; +import org.apache.iceberg.mr.hive.test.TestTables.TestTableType; +import org.apache.iceberg.mr.mapreduce.IcebergSplitContainer; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Streams; +import org.apache.iceberg.types.Types; +import org.junit.Assume; +import org.junit.Test; +import org.junit.runners.Parameterized.Parameters; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * File-level pruning of shredded VARIANT values during split planning. + */ +public class TestHiveIcebergVariantFilePruning extends HiveIcebergStorageHandlerWithEngineBase { + + private static final Schema SCHEMA = new Schema( + required(1, "id", Types.IntegerType.get()), + required(2, "region", Types.StringType.get()), + optional(3, "payload", Types.VariantType.get())); + + @Parameters(name = "fileFormat={0}, catalog={1}, isVectorized={2}, formatVersion={3}") + public static Collection parameters() { + return HiveIcebergStorageHandlerWithEngineBase.getParameters(p -> + p.fileFormat() == FileFormat.PARQUET && + p.testTableType() == TestTableType.HIVE_CATALOG && + p.formatVersion() == 3); + } + + @Test + public void testSelectivePredicatePlansOnlyMatchingFile() throws Exception { + // split planning does not depend on the execution mode + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning"; + createTable(table, PartitionSpec.unpartitioned(), true); + insert(table, events(1, 10, "eu")); + String second = insert(table, events(11, 20, "eu")); + insert(table, events(21, 30, "eu")); + + assertThat(plannedFiles(table, eq(eventId("int"), intLit(15)))) + .containsExactly(second); + } + + @Test + public void testExtractPredicatesAndTheirNegationsPrune() throws Exception { + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning_conjuncts"; + createTable(table, PartitionSpec.unpartitioned(), true); + String first = insert(table, events(1, 10, "eu")); + String second = insert(table, events(11, 20, "eu")); + String third = insert(table, events(21, 30, "eu")); + List all = sorted(first, second, third); + + assertThat(plannedFiles(table, and(gt(id(), intLit(0)), eq(eventId("int"), intLit(15))))) + .containsExactly(second); + assertThat(plannedFiles(table, lt(eventId("int"), intLit(11)))) + .containsExactly(first); + assertThat(plannedFiles(table, lt(eventId("int"), intLit(100)))) + .isEqualTo(all); + assertThat(plannedFiles(table, between(false, eventId("int"), intLit(12), intLit(14)))) + .containsExactly(second); + assertThat(plannedFiles(table, in(eventId("int"), intLit(15), intLit(16)))) + .containsExactly(second); + assertThat(plannedFiles(table, in(eventId("int"), intLit(5), intLit(25)))) + .isEqualTo(sorted(first, third)); + assertThat(plannedFiles(table, or(eq(eventId("int"), intLit(15)), eq(eventId("int"), intLit(25))))) + .isEqualTo(all); + assertThat(plannedFiles(table, isNull(eventId("int")))) + .isEqualTo(all); + + // Hive's search argument writes >, >=, <> and IS NOT NULL as NOT of <=, <, = and IS NULL + assertThat(plannedFiles(table, gt(eventId("int"), intLit(25)))) + .containsExactly(third); + assertThat(plannedFiles(table, gtEq(eventId("int"), intLit(20)))) + .isEqualTo(sorted(second, third)); + assertThat(plannedFiles(table, not(lt(eventId("int"), intLit(11))))) + .isEqualTo(sorted(second, third)); + // a file can be dropped only if its one value is 15, or in the list + assertThat(plannedFiles(table, notEq(eventId("int"), intLit(15)))) + .isEqualTo(all); + assertThat(plannedFiles(table, not(in(eventId("int"), intLit(15), intLit(16))))) + .isEqualTo(all); + // only the variant column's null count is known, and no file holds only NULL variants + assertThat(plannedFiles(table, isNotNull(eventId("int")))) + .isEqualTo(all); + // the search argument pushes NOT below OR: NOT(a OR b) is NOT a AND NOT b + assertThat(plannedFiles(table, not(or(lt(eventId("int"), intLit(11)), eq(eventId("int"), intLit(25)))))) + .isEqualTo(sorted(second, third)); + // ... and NOT(a AND b) becomes an OR, which is ignored, as is NOT over a BETWEEN + assertThat(plannedFiles(table, not(and(lt(eventId("int"), intLit(11)), gt(id(), intLit(0)))))) + .isEqualTo(all); + assertThat(plannedFiles(table, between(true, eventId("int"), intLit(1), intLit(10)))) + .isEqualTo(all); + + // nothing pruned: the files are combined into the same splits as without the pruner + assertThat(plannedSplits(table, lt(eventId("int"), intLit(100)))) + .isEqualTo(plannedSplits(table, lt(id(), intLit(100)))) + .hasSize(1); + } + + @Test + public void testIsNotNullDropsFileOfNullVariants() throws Exception { + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning_null_variants"; + createTable(table, PartitionSpec.unpartitioned(), true); + String events = insert(table, events(1, 10, "eu")); + String nulls = insertInto(table, "(id, region)", "(11, 'eu'), (12, 'eu')"); + + assertThat(plannedFiles(table, isNotNull(eventId("int")))) + .containsExactly(events); + assertThat(plannedFiles(table, isNull(eventId("int")))) + .isEqualTo(sorted(events, nulls)); + } + + @Test + public void testFilesWithoutVariantBoundsAreKept() throws Exception { + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning_no_bounds"; + createTable(table, PartitionSpec.unpartitioned(), true); + String shredded = insert(table, events(1, 10, "eu")); + String unshredded = insertUnshredded(table, "(11, 'eu', parse_json('{\"event_id\": 15}'))"); + String fallback = insert(table, + "(21, 'eu', parse_json('{\"event_id\": 41}')), (22, 'eu', parse_json('{\"event_id\": \"x\"}'))"); + + assertThat(filesWithVariantBounds(table)).containsExactly(shredded); + assertThat(plannedFiles(table, eq(eventId("int"), intLit(15)))) + .isEqualTo(sorted(unshredded, fallback)); + } + + @Test + public void testLossyCastKeepsFile() throws Exception { + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning_casts"; + createTable(table, PartitionSpec.unpartitioned(), true); + String ints = insert(table, events(1, 10, "eu")); + String doubles = insert(table, "(11, 'eu', parse_json('{\"event_id\": 1.57e1}'))"); + String decimals = insert(table, "(12, 'eu', parse_json('{\"event_id\": 15.70}'))"); + String strings = insert(table, "(13, 'eu', parse_json('{\"event_id\": \"15\"}'))"); + String moreInts = insert(table, events(14, 20, "eu")); + + assertThat(filesWithVariantBounds(table)).containsExactlyInAnyOrder(ints, doubles, decimals, strings, moreInts); + // Hive casts 15.7, 15.70 and "15" to the int 15 + assertThat(plannedFiles(table, eq(eventId("int"), intLit(15)))) + .isEqualTo(sorted(doubles, decimals, strings, moreInts)); + assertThat(plannedFiles(table, eq(eventId("double"), doubleLit(15.0)))) + .isEqualTo(sorted(ints, decimals, strings, moreInts)); + assertThat(plannedFiles(table, eq(eventId("string"), stringLit("16")))) + .isEqualTo(sorted(ints, doubles, decimals, moreInts)); + } + + @Test + public void testPartitionedTable() throws Exception { + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning_partitioned"; + createTable(table, PartitionSpec.builderFor(SCHEMA).identity("region").build(), true); + String first = insert(table, events(1, 10, "eu")); + String second = insert(table, events(11, 20, "us")); + insert(table, events(21, 30, "eu")); + + // grouping by "region" plans files and groups them, instead of planning tasks + List> paths = + List.of(job -> { }, job -> job.set(TableScanDesc.GROUPING_PARTITION_COLUMNS, "region")); + for (Consumer path : paths) { + assertThat(plannedFiles(table, path, eq(eventId("int"), intLit(15)))) + .containsExactly(second); + assertThat(plannedFiles(table, path, lt(eventId("int"), intLit(11)))) + .containsExactly(first); + } + } + + @Test + public void testIncrementalAppendScan() throws Exception { + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning_incremental"; + createTable(table, PartitionSpec.unpartitioned(), true); + insert(table, events(1, 10, "eu")); + long first = testTables.loadTable(TableIdentifier.of("default", table)).currentSnapshot().snapshotId(); + String second = insert(table, events(11, 20, "eu")); + String third = insert(table, events(21, 30, "eu")); + Consumer appendedAfterFirst = job -> job.set(TableScanDesc.FROM_VERSION, String.valueOf(first)); + + assertThat(plannedFiles(table, appendedAfterFirst, lt(id(), intLit(100)))) + .isEqualTo(sorted(second, third)); + assertThat(plannedFiles(table, appendedAfterFirst, eq(eventId("int"), intLit(15)))) + .containsExactly(second); + assertThat(plannedFiles(table, appendedAfterFirst, eq(eventId("int"), intLit(5)))) + .isEmpty(); + } + + @Test + public void testNestedPath() throws Exception { + String shredded = "variant_file_pruning_nested"; + String plain = "variant_file_pruning_nested_plain"; + createTable(shredded, PartitionSpec.unpartitioned(), true); + createTable(plain, PartitionSpec.unpartitioned(), false); + String smallRows = "(1, 'eu', parse_json('{\"a\": {\"b\": 5}}')), (2, 'eu', parse_json('{\"a\": {\"b\": 6}}'))"; + String largeRows = "(3, 'eu', parse_json('{\"a\": {\"b\": 50}}'))"; + String small = insert(shredded, smallRows); + String large = insert(shredded, largeRows); + insert(plain, smallRows); + insert(plain, largeRows); + + ExprNodeDesc ab = variantGet("$.a.b", "int"); + assertThat(plannedFiles(shredded, eq(ab, intLit(50)))).containsExactly(large); + assertThat(plannedFiles(shredded, lt(ab, intLit(10)))).containsExactly(small); + + String predicate = "variant_get(payload, '$.a.b', 'int')"; + for (String query : List.of(predicate + " = 50", predicate + " = 5", predicate + " > 5")) { + assertThat(ids(shredded, query)).as(query).isEqualTo(ids(plain, query)); + } + assertThat(ids(shredded, predicate + " = 50")).containsExactly(3); + } + + @Test + public void testQueryReadsOnlyMatchingFiles() throws Exception { + String table = "variant_file_pruning_missing_files"; + createTable(table, PartitionSpec.unpartitioned(), true); + String first = insert(table, events(1, 10, "eu")); + insert(table, events(11, 20, "eu")); + String third = insert(table, events(21, 30, "eu")); + // a query that plans a non-matching file now fails reading it + assertThat(new File(new URI(first)).delete()).isTrue(); + assertThat(new File(new URI(third)).delete()).isTrue(); + + String eventId = "variant_get(payload, '$.event_id', 'int')"; + assertThat(ids(table, eventId + " = 15")).containsExactly(15); + assertThat(ids(table, eventId + " IN (15, 16)")).containsExactly(15, 16); + assertThat(ids(table, eventId + " > 12 AND " + eventId + " < 14")).containsExactly(13); + assertThatThrownBy(() -> ids(table, eventId + " = 15 OR id < 0")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testResultsMatchUnshreddedTable() { + String shredded = "variant_results_shredded"; + String plain = "variant_results_plain"; + createTable(shredded, PartitionSpec.unpartitioned(), true); + createTable(plain, PartitionSpec.unpartitioned(), false); + for (String table : List.of(shredded, plain)) { + insert(table, events(1, 10, "eu")); + insert(table, events(11, 20, "eu")); + insert(table, events(21, 30, "eu")); + String unshreddedRows = events(31, 32, "eu") + ", (33, 'eu', parse_json('{\"event_id\": 15}'))"; + if (table.equals(shredded)) { + insertUnshredded(table, unshreddedRows); + } else { + insert(table, unshreddedRows); + } + insert(table, "(41, 'eu', parse_json('{\"event_id\": 41}')), " + + "(42, 'eu', parse_json('{\"event_id\": \"x\"}')), (43, 'eu', parse_json('{\"event_id\": \"15\"}'))"); + insert(table, "(51, 'eu', parse_json('{\"kind\": \"z\"}')), (52, 'eu', parse_json('null'))"); + shell.executeStatement(String.format("INSERT INTO default.%s (id, region) VALUES (53, 'eu')", table)); + insert(table, + "(61, 'eu', parse_json('{\"event_id\": \"15\"}')), (62, 'eu', parse_json('{\"event_id\": \"16\"}'))"); + } + + String eventId = "variant_get(payload, '$.event_id', 'int')"; + assertThat(ids(shredded, eventId + " = 15")).containsExactly(15, 33, 43, 61); + List predicates = List.of( + eventId + " = 15", + eventId + " < 11", + eventId + " > 0", + eventId + " = 15 OR " + eventId + " = 25", + "NOT (" + eventId + " < 11)", + eventId + " > 25", + eventId + " >= 20", + eventId + " <> 15", + eventId + " IN (15, 16, 25)", + eventId + " NOT IN (15, 16)", + "variant_get(payload, '$.kind', 'string') IS NULL", + "variant_get(payload, '$.kind', 'string') IS NOT NULL", + "id > 5 AND " + eventId + " BETWEEN 12 AND 14", + "variant_get(payload, '$.event_id', 'string') = '16'", + "variant_get(payload, '$.event_id', 'double') = 15.0"); + for (String predicate : predicates) { + assertThat(ids(shredded, predicate)).as(predicate).isEqualTo(ids(plain, predicate)); + } + } + + @Test + public void testTimeTravelBeforeVariantColumnExisted() { + Assume.assumeFalse(isVectorized); + String table = "variant_file_pruning_time_travel"; + testTables.createTable(shell, table, new Schema(required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), fileFormat, ImmutableList.of(), formatVersion, + ImmutableMap.of(InputFormatConfig.VARIANT_SHREDDING_ENABLED, "true")); + shell.executeStatement(String.format("INSERT INTO default.%s VALUES (1)", table)); + long before = testTables.loadTable(TableIdentifier.of("default", table)).currentSnapshot().snapshotId(); + shell.executeStatement(String.format("ALTER TABLE default.%s ADD COLUMNS (payload variant)", table)); + shell.executeStatement( + String.format("INSERT INTO default.%s VALUES (2, parse_json('{\"event_id\": 15}'))", table)); + + String predicate = "variant_get(payload, '$.event_id', 'int') = 15"; + assertThat(ids(table, predicate)).containsExactly(2); + assertThat(ids(table + " FOR SYSTEM_VERSION AS OF " + before, predicate)).isEmpty(); + } + + private void createTable(String table, PartitionSpec spec, boolean shredded) { + testTables.createTable(shell, table, SCHEMA, spec, fileFormat, ImmutableList.of(), formatVersion, + ImmutableMap.of(InputFormatConfig.VARIANT_SHREDDING_ENABLED, String.valueOf(shredded))); + } + + /** + * Inserts the rows in one statement and returns the location of the single data file it adds. + */ + private String insert(String table, String rows) { + return insertInto(table, "", rows); + } + + private String insertInto(String table, String columns, String rows) { + shell.executeStatement(String.format("INSERT INTO default.%s %s VALUES %s", table, columns, rows)); + Table icebergTable = testTables.loadTable(TableIdentifier.of("default", table)); + List added = Lists.newArrayList(icebergTable.currentSnapshot().addedDataFiles(icebergTable.io())); + assertThat(added).hasSize(1); + return added.getFirst().location(); + } + + /** + * Inserts into a shredded table with shredding switched off for the insert. + */ + private String insertUnshredded(String table, String rows) { + setShredding(table, false); + String file = insert(table, rows); + setShredding(table, true); + return file; + } + + private void setShredding(String table, boolean enabled) { + shell.executeStatement(String.format("ALTER TABLE default.%s SET TBLPROPERTIES ('%s'='%s')", + table, InputFormatConfig.VARIANT_SHREDDING_ENABLED, enabled)); + } + + private static String events(int from, int to, String region) { + return IntStream.rangeClosed(from, to) + .mapToObj(id -> String.format("(%d, '%s', parse_json('{\"event_id\": %d}'))", id, region, id)) + .collect(Collectors.joining(", ")); + } + + private List ids(String table, String predicate) { + return shell.executeStatement(String.format("SELECT id FROM default.%s WHERE %s ORDER BY id", table, predicate)) + .stream() + .map(row -> ((Number) row[0]).intValue()) + .toList(); + } + + private Set filesWithVariantBounds(String table) throws IOException { + Table icebergTable = testTables.loadTable(TableIdentifier.of("default", table)); + int payloadId = icebergTable.schema().findField("payload").fieldId(); + try (CloseableIterable tasks = icebergTable.newScan().includeColumnStats().planFiles()) { + return Streams.stream(tasks) + .map(FileScanTask::file) + .filter(file -> file.lowerBounds() != null && file.lowerBounds().containsKey(payloadId)) + .map(DataFile::location) + .collect(Collectors.toSet()); + } + } + + private List plannedFiles(String table, ExprNodeGenericFuncDesc filter) throws Exception { + return plannedFiles(table, job -> { }, filter); + } + + private List plannedFiles(String table, Consumer settings, ExprNodeGenericFuncDesc filter) + throws Exception { + return sorted(plannedSplits(table, settings, filter).stream() + .flatMap(List::stream) + .toArray(String[]::new)); + } + + private List> plannedSplits(String table, ExprNodeGenericFuncDesc filter) throws Exception { + return plannedSplits(table, job -> { }, filter); + } + + /** + * Plans splits the way Hive does for a table scan: job properties from the storage handler, the pushed-down Hive + * predicate and the given scan settings. Returns the sorted data file locations of each split. + */ + private List> plannedSplits(String table, Consumer settings, + ExprNodeGenericFuncDesc filter) throws Exception { + TableDesc tableDesc = Utilities.getTableDesc( + new org.apache.hadoop.hive.ql.metadata.Table(shell.metastore().getTable("default", table))); + HiveIcebergStorageHandler storageHandler = new HiveIcebergStorageHandler(); + storageHandler.setConf(shell.getHiveConf()); + Map jobProperties = Maps.newHashMap(); + storageHandler.configureInputJobProperties(tableDesc, jobProperties); + tableDesc.setJobProperties(jobProperties); + + JobConf job = new JobConf(shell.getHiveConf()); + Utilities.copyTableJobPropertiesToConf(tableDesc, job); + job.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(filter)); + settings.accept(job); + + InputSplit[] splits = new HiveIcebergInputFormat().getSplits(job, 0); + return Arrays.stream(splits) + .map(split -> sorted(((IcebergSplitContainer) split).icebergSplit().taskGroup().tasks().stream() + .map(task -> task.file().location()) + .toArray(String[]::new))) + .sorted(Comparator.comparing(List::getFirst)) + .toList(); + } + + private static List sorted(String... files) { + return Arrays.stream(files).sorted().toList(); + } + + private static ExprNodeDesc id() { + return new ExprNodeColumnDesc(TypeInfoFactory.intTypeInfo, "id", "t", false); + } + + private static ExprNodeDesc eventId(String type) { + return variantGet("$.event_id", type); + } + + private static ExprNodeDesc variantGet(String path, String type) { + return func(new GenericUDFVariantGet(), + new ExprNodeColumnDesc(TypeInfoFactory.getVariantTypeInfo(), "payload", "t", false), + new ExprNodeConstantDesc(TypeInfoFactory.stringTypeInfo, path), + new ExprNodeConstantDesc(TypeInfoFactory.stringTypeInfo, type)); + } + + private static ExprNodeConstantDesc intLit(int value) { + return new ExprNodeConstantDesc(TypeInfoFactory.intTypeInfo, value); + } + + private static ExprNodeConstantDesc doubleLit(double value) { + return new ExprNodeConstantDesc(TypeInfoFactory.doubleTypeInfo, value); + } + + private static ExprNodeConstantDesc stringLit(String value) { + return new ExprNodeConstantDesc(TypeInfoFactory.stringTypeInfo, value); + } + + private static ExprNodeGenericFuncDesc eq(ExprNodeDesc left, ExprNodeDesc right) { + return func(new GenericUDFOPEqual(), left, right); + } + + private static ExprNodeGenericFuncDesc notEq(ExprNodeDesc left, ExprNodeDesc right) { + return func(new GenericUDFOPNotEqual(), left, right); + } + + private static ExprNodeGenericFuncDesc lt(ExprNodeDesc left, ExprNodeDesc right) { + return func(new GenericUDFOPLessThan(), left, right); + } + + private static ExprNodeGenericFuncDesc gt(ExprNodeDesc left, ExprNodeDesc right) { + return func(new GenericUDFOPGreaterThan(), left, right); + } + + private static ExprNodeGenericFuncDesc gtEq(ExprNodeDesc left, ExprNodeDesc right) { + return func(new GenericUDFOPEqualOrGreaterThan(), left, right); + } + + private static ExprNodeGenericFuncDesc between(boolean invert, ExprNodeDesc expr, ExprNodeDesc low, + ExprNodeDesc high) { + return func(new GenericUDFBetween(), + new ExprNodeConstantDesc(TypeInfoFactory.booleanTypeInfo, invert), expr, low, high); + } + + private static ExprNodeGenericFuncDesc and(ExprNodeDesc left, ExprNodeDesc right) { + return func(new GenericUDFOPAnd(), left, right); + } + + private static ExprNodeGenericFuncDesc or(ExprNodeDesc left, ExprNodeDesc right) { + return func(new GenericUDFOPOr(), left, right); + } + + private static ExprNodeGenericFuncDesc not(ExprNodeDesc child) { + return func(new GenericUDFOPNot(), child); + } + + private static ExprNodeGenericFuncDesc isNull(ExprNodeDesc child) { + return func(new GenericUDFOPNull(), child); + } + + private static ExprNodeGenericFuncDesc in(ExprNodeDesc expr, ExprNodeDesc... values) { + ExprNodeDesc[] children = new ExprNodeDesc[values.length + 1]; + children[0] = expr; + System.arraycopy(values, 0, children, 1, values.length); + return func(new GenericUDFIn(), children); + } + + private static ExprNodeGenericFuncDesc isNotNull(ExprNodeDesc child) { + return func(new GenericUDFOPNotNull(), child); + } + + private static ExprNodeGenericFuncDesc func(GenericUDF udf, ExprNodeDesc... children) { + try { + return ExprNodeGenericFuncDesc.newInstance(udf, Lists.newArrayList(children)); + } catch (Exception e) { + throw new IllegalArgumentException(e); + } + } +} diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/variant/TestVariantFilePruner.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/variant/TestVariantFilePruner.java new file mode 100644 index 000000000000..45808006feec --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/variant/TestVariantFilePruner.java @@ -0,0 +1,157 @@ +/* + * 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 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.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.PathUtil; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ByteBuffers; +import org.apache.iceberg.variants.ShreddedObject; +import org.apache.iceberg.variants.Variant; +import org.apache.iceberg.variants.VariantMetadata; +import org.apache.iceberg.variants.Variants; +import org.junit.Test; +import org.mockito.Mockito; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; + +public class TestVariantFilePruner { + + private static final Schema SCHEMA = new Schema( + required(1, "id", Types.IntegerType.get()), + optional(2, "payload", Types.VariantType.get())); + + private static final Expression EVENT_ID_15 = + Expressions.equal(Expressions.extract("payload", "$.event_id", "long"), 15L); + + @Test + public void testNoPrunerWithoutTopLevelExtractConjunct() { + assertThat(VariantFilePruner.of(SCHEMA, null, true)).isNull(); + assertThat(VariantFilePruner.of(SCHEMA, Expressions.equal("id", 5), true)).isNull(); + assertThat(VariantFilePruner.of(SCHEMA, Expressions.or(EVENT_ID_15, Expressions.equal("id", 5)), true)).isNull(); + assertThat(VariantFilePruner.of(SCHEMA, Expressions.not(Expressions.and(EVENT_ID_15, EVENT_ID_15)), true)) + .isNull(); + assertThat(VariantFilePruner.of(SCHEMA, Expressions.not(Expressions.or(EVENT_ID_15, EVENT_ID_15)), true)) + .isNull(); + assertThat(VariantFilePruner.of(SCHEMA, Expressions.not(Expressions.equal("id", 5)), true)).isNull(); + // DECIMAL fallback of HiveIcebergFilterFactory, cannot bind to the table schema + assertThat(VariantFilePruner.of(SCHEMA, Expressions.equal("payload.typed_value.price", 5), true)).isNull(); + // a time-travel scan before the variant column was added + assertThat(VariantFilePruner.of(SCHEMA.select("id"), EVENT_ID_15, true)).isNull(); + } + + @Test + public void testNoPrunerForExtractOnNonVariantColumn() { + // variant_get accepts any struct column + Schema schema = new Schema( + required(1, "id", Types.IntegerType.get()), + optional(2, "s", Types.StructType.of( + optional(3, "metadata", Types.BinaryType.get()), + optional(4, "value", Types.BinaryType.get())))); + Expression onStruct = Expressions.equal(Expressions.extract("s", "$.x", "long"), 1L); + + assertThat(VariantFilePruner.of(schema, onStruct, true)).isNull(); + assertThat(VariantFilePruner.of(schema, Expressions.not(onStruct), true)).isNull(); + } + + @Test + public void testCaseInsensitiveColumn() { + Expression upperCase = Expressions.equal(Expressions.extract("PAYLOAD", "$.event_id", "long"), 15L); + assertThat(VariantFilePruner.of(SCHEMA, upperCase, true)).isNull(); + assertThat(VariantFilePruner.of(SCHEMA, upperCase, false).variantColumns()).containsExactly("payload"); + } + + @Test + public void testPrunesWithManifestBounds() { + ByteBuffer lower = manifestBound(1); + ByteBuffer upper = manifestBound(10); + FileScanTask task = task(lower, upper); + + VariantFilePruner pruner = + VariantFilePruner.of(SCHEMA, Expressions.and(Expressions.equal("id", 5), EVENT_ID_15), true); + assertThat(pruner.variantColumns()).containsExactly("payload"); + assertThat(pruner.mayMatch(task)).isFalse(); + + Expression eventId5 = Expressions.equal(Expressions.extract("payload", "$.event_id", "long"), 5L); + assertThat(VariantFilePruner.of(SCHEMA, eventId5, true).mayMatch(task)).isTrue(); + + assertThat(lower.order()).isEqualTo(ByteOrder.BIG_ENDIAN); + assertThat(lower.position()).isZero(); + assertThat(task.file().lowerBounds().get(2)).isSameAs(lower); + } + + @Test + public void testNegatedExtractPredicate() { + FileScanTask task = task(manifestBound(1), manifestBound(10)); + Expression notAtMost10 = + Expressions.not(Expressions.lessThanOrEqual(Expressions.extract("payload", "$.event_id", "long"), 10L)); + Expression notAtMost9 = + Expressions.not(Expressions.lessThanOrEqual(Expressions.extract("payload", "$.event_id", "long"), 9L)); + + assertThat(VariantFilePruner.of(SCHEMA, notAtMost10, true).mayMatch(task)).isFalse(); + assertThat(VariantFilePruner.of(SCHEMA, notAtMost9, true).mayMatch(task)).isTrue(); + } + + @Test + public void testKeepsFileWithoutVariantBounds() { + assertThat(VariantFilePruner.of(SCHEMA, EVENT_ID_15, true).mayMatch(task(null, null))).isTrue(); + } + + /** + * A variant bound for {@code $.event_id} as a manifest returns it: in a big-endian buffer. + */ + private static ByteBuffer manifestBound(int eventId) { + String path = PathUtil.toNormalizedPath(List.of("event_id")); + VariantMetadata metadata = Variants.metadata(path); + ShreddedObject bound = Variants.object(metadata); + bound.put(path, Variants.of((byte) eventId)); + ByteBuffer serialized = Conversions.toByteBuffer(Types.VariantType.get(), Variant.of(metadata, bound)); + return ByteBuffer.wrap(ByteBuffers.toByteArray(serialized)); + } + + private static FileScanTask task(ByteBuffer lower, ByteBuffer upper) { + DataFile file = DataFiles.builder(PartitionSpec.unpartitioned()) + .withPath("file:/tmp/data.parquet") + .withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(100L) + .withMetrics(new Metrics(10L, null, Map.of(2, 10L), Map.of(2, 0L), null, + lower == null ? null : Map.of(2, lower), upper == null ? null : Map.of(2, upper))) + .build(); + FileScanTask task = Mockito.mock(FileScanTask.class); + Mockito.when(task.file()).thenReturn(file); + Mockito.when(task.spec()).thenReturn(PartitionSpec.unpartitioned()); + return task; + } +}