You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "steveloughran (via GitHub)" <gi...@apache.org> on 2026/03/13 21:25:57 UTC
[PR] Add JMH benchmarks for Variants [iceberg]
steveloughran opened a new pull request, #15629:
URL: https://github.com/apache/iceberg/pull/15629
Fixes #15628
Separate benchmarks for
* building
* serializing a prebuilt object
* deserializing
Variables are:
- fields: [1000, 10000]
- depth: [shallow, nested]
- percentage of fields shed [0, 33, 67, 100]
Note: the current benchmark does NOT for the JVM, as it allows for fast iterative development. A final merge should switch to fork(1).
The benchmarks don't show any surprises, which is good
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3588117673
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,915 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RuntimeConfig;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant object
+ * <li>fully shredded variant object
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in a very small range; written in clustered order so each Parquet row group should
+ * contain very few entries from different ranges.
+ * nested: variant (object)
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32
+ * .col4: string :- non-unique string per row (same number as category count)
+ * arr: variant (array, always unshredded)
+ * [0]: int32 :- category
+ * [1]: int32 :- id % 20
+ * [2]: int32 :- id % 100
+ * [3]: int32 :- id % 50
+ * [4]: int32 :- id % 1000
+ * </pre>
+ *
+ * <p>Arrays don't shred because they are anonymous.
+ *
+ * <p>All data is written into a single file to keep the Spark task count to 1. Combined with a
+ * {@code local[1]} Spark master, all benchmark iterations execute on a single thread, which
+ * eliminates inter-task scheduling noise and makes measurements reproducible.
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shredded
+ * parquet, irrespective of performance — look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 5)
+@Measurement(iterations = 5)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 60, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ private static final int NUM_FILES = 1;
+ private static final int NUM_ROWS_PER_FILE = 100_000;
+
+ /** Size of a compressed rowgroup. */
+ private static final int ROW_GROUP_SIZE = 1 * 1024 * 1024;
+
+ /** Number of distinct category values. */
+ private static final int NUM_CATEGORIES = 10;
+
+ public static final String COL_ID = "id";
+ private static final String COL_NESTED = "nested";
+ private static final String COL_ARR = "arr";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ /**
+ * Name of the second Spark temp view for the self-join benchmark. Both views read the same
+ * underlying table; registering two aliases avoids any Spark self-join restrictions.
+ */
+ private static final String TEMP_VIEW2 = "variant_table2";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, COL_NESTED, Types.VariantType.get()), /* The variant object. */
+ required(4, COL_ARR, Types.VariantType.get())); /* The variant array. */
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'bigint')";
+
+ /** Equality check. Pulled out to allow experimentation with set membershop vs. equals. */
+ public static final String EQUALITY_CHECK = " = 1";
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category " + EQUALITY_CHECK;
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}. */
+ private static final String FILTER_ON_NESTED_CATEGORY =
+ VARIANT_GET_NESTED_CATEGORY + EQUALITY_CHECK;
+
+ /**
+ * Get element 0 of the array variant: {@value}.
+ *
+ * <p>Note: {@code variant_get} with a JSONPath {@code $[n]} expression accesses array elements.
+ */
+ private static final String VARIANT_GET_ARR_ELT0 = "variant_get(arr, '$[0]', 'int')";
+
+ /** Filter on element 0 of the array variant: {@value}. */
+ private static final String FILTER_ON_ARR_ELT0 = VARIANT_GET_ARR_ELT0 + EQUALITY_CHECK;
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * Use a single-threaded Spark master to eliminate inter-task scheduling noise.
+ *
+ * <p>With one file and {@code local[1]}, every scan benchmark runs as a single serial task.
+ */
+ @Override
+ protected String sparkMaster() {
+ return "local[1]";
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // large split size keeps the single data file as one Spark task.
+ properties.put(TableProperties.SPLIT_SIZE, Integer.toString(512 * 1024 * 1024));
+ // small row group size ensures multiple row groups per file for row-group skipping benchmarks.
+ properties.put(TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES, Integer.toString(ROW_GROUP_SIZE));
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup(Level.Trial)
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+ final RuntimeConfig sc = spark().conf();
+ // one shuffle partition keeps the join benchmark on a single task.
+ sc.set("spark.sql.shuffle.partitions", "1");
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ isShredded()
+ ? (fieldId, fieldName) -> COL_NESTED.equals(fieldName) ? shreddedType : null
+ : (fieldId, fieldName) -> null;
+
+ final int categoryCount = NUM_CATEGORIES;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ tableDataset().createOrReplaceTempView(TEMP_VIEW2);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ logRowGroupFiltering();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /**
+ * Is the table shredded?
+ *
+ * @return true if the table is shredded.
+ */
+ private boolean isShredded() {
+ return tableType.shredded;
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count(Blackhole blackhole) {
+ project(blackhole, "*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_CATEGORY, false);
+ }
+
+ /**
+ * Perform a sql select operation to non-zero count of rows.
+ *
+ * @param blackhole consumer of all data
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @param expectPushdownWhenShredded on a shredded table, should the rowgroup filter report
+ * invocation?
+ * @return the result.
+ */
+ private long select(
+ final Blackhole blackhole,
+ String projection,
+ String where,
+ boolean expectPushdownWhenShredded) {
+ final long result = sql(blackhole, toSql(projection, where));
+ if (expectPushdownWhenShredded) {
+ expectFilteringOfShreddedFields();
+ }
+ return result;
+ }
+
+ /**
+ * Perform a sql select operation where the result is an empty set.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @param expectPushdownWhenShredded on a shredded table, should the rowgroup filter report
+ * invocation?
+ */
+ private void selectEmptyResults(
+ final String projection, final String where, boolean expectPushdownWhenShredded) {
+ String command = toSql(projection, where);
+ resetMetricsCounter();
+ final Dataset<Row> ds = spark().sql(command);
+ assertThat(ds.collectAsList()).describedAs("Result of command %s", command).isEmpty();
+ if (expectPushdownWhenShredded) {
+ expectFilteringOfShreddedFields();
+ }
+ }
+
+ /**
+ * Create a SQL select statement.
+ *
+ * @param projection projection clause
+ * @param where filter, may be null
+ * @return the string for evaluation.
+ */
+ private static String toSql(final String projection, final String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ return text;
+ }
+
+ /**
+ * Issue any Spark SQL command that returns a non-zero count of rows.
+ *
+ * @param blackhole consumer of all data
+ * @param command command to execute
+ * @return count of result.
+ */
+ private long sql(final Blackhole blackhole, String command) {
+ LOG.info("{}", command);
+ resetMetricsCounter();
+ return materializeNonEmpty(blackhole, command, spark().sql(command));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(final Blackhole blackhole, String projection) {
+ return select(blackhole, projection, null, false);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY, false);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID(Blackhole blackhole) {
+ project(blackhole, COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat(Blackhole blackhole) {
+ select(blackhole, "*", FILTER_ON_CATEGORY, false);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat(Blackhole blackhole) {
+ select(blackhole, "*", FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ @Benchmark
+ public void filterVarCatSetMembership(Blackhole blackhole) {
+ select(blackhole, COL_ID, VARIANT_GET_NESTED_CATEGORY + " IN (1)", true);
+ }
+
+ @Benchmark
+ public void filterVarCatSetNotInMembership(Blackhole blackhole) {
+ selectEmptyResults(COL_ID, VARIANT_GET_NESTED_CATEGORY + " IN (100, 400)", false);
+ }
+
+ @Benchmark
+ public void filterVarCatSetNotInRange(Blackhole blackhole) {
+ selectEmptyResults(COL_ID, VARIANT_GET_NESTED_CATEGORY + " < 0", false);
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ /** selecting on 1, but using a pair of ranges to locate. Two scans needed */
+ @Benchmark
+ public void filterVarcatProjectVarIDRanged(Blackhole blackhole) {
+ select(
+ blackhole,
+ VARIANT_GET_NESTED_ID,
+ VARIANT_GET_NESTED_CATEGORY + " > 0 and " + VARIANT_GET_NESTED_CATEGORY + " < 2",
+ true);
+ }
+
+ /**
+ * Project element 0 of the array variant.
+ *
+ * <p>Note: the {@code arr} column is stored unshredded in all table types (array shredding is out
+ * of scope). This benchmark compares unshredded Parquet vs Avro array element access.
+ */
+ @Benchmark
+ public void projectArrayCategory(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_ARR_ELT0);
+ }
+
+ /** Filter on element 0 of the array variant then project on ID. */
+ @Benchmark
+ public void filterArrayCategoryProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_ARR_ELT0, false);
+ }
+
+ /**
+ * Filter on element 0 of the array variant then project element 0.
+ *
+ * <p>Both the filter and projection touch the same array element; this measures the cost of
+ * evaluating {@code variant_get} on an array twice per row (once for filter, once for output).
+ */
+ @Benchmark
+ public void filterArray0ProjectArray0(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_ARR_ELT0, FILTER_ON_ARR_ELT0, false);
+ }
+
+ /** Explode the array variant into one row per element. */
+ @Benchmark
+ public void explodeArraryColumns(Blackhole blackhole) {
+ sql(blackhole, "select t.* from " + TEMP_VIEW + " as table, lateral variant_explode(arr) as t");
+ }
+
+ /**
+ * Self-join the table on {@code arr[4]} (= {@code id % 1000}, range 0-999) against {@code
+ * varcategory} from the nested variant.
+ *
+ * <p>TODO: failing in spark. Needs investigation.
+ */
+ // @Benchmark
+ public void joinArrElt4OnVarcat(Blackhole blackhole) {
+ final String text =
+ "SELECT t1."
+ + COL_ID
+ + ", t2."
+ + COL_ID
+ + " FROM "
+ + TEMP_VIEW
+ + " t1 JOIN "
+ + TEMP_VIEW2
+ + " t2 ON "
+ + "variant_get(t1.arr, '$[4]', 'int') = variant_get(t2.nested, '$.varcategory', 'int')";
+ sql(blackhole, text);
+ expectFilteringOfShreddedFields();
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param blackhole consumer of all data
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(final Blackhole blackhole, String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.queryExecution().toRdd().toJavaRDD().count();
+ blackhole.consume(count);
+ Preconditions.checkState(
+ count > 0, "Operation %s on %s table returned no results", operation, tableType);
+ LOG.info("Materialization of Dataset returned {} records", count);
+ return count;
+ }
+
+ /**
+ * Generate files in the target subdir, sorted by timestamp and with their format included in the
+ * path.
+ *
+ * @return the location for the new table
+ */
+ @Override
+ protected String newTableLocation() {
+ Path tablePath =
+ new Path(
+ String.format(
+ Locale.ROOT,
+ "build/tables/spark-iceberg-table-%d-%s",
+ System.currentTimeMillis(),
+ UUID.randomUUID()));
+ return tablePath.toString();
+ }
+
+ /**
+ * Build the parquet dataset.
+ *
+ * @throws IOException creation failure.
+ */
+ private long appendVariantData() throws IOException {
+ long size = 0;
+ final AppendFiles append = table().newAppend();
+
+ for (int fileNum = 0; fileNum < NUM_FILES; fileNum++) {
+ final DataFile dataFile = createOneParquetFile(fileNum);
+ append.appendFile(dataFile);
+ size += dataFile.fileSizeInBytes();
+ }
+ append.commit();
+ return size;
+ }
+
+ /**
+ * Create a single parquet file as part of the append operation.
+ *
+ * @param fileIndex file index to use in generation of name and category derivation.
+ * @return the data file.
+ * @throws IOException write failure.
+ */
+ private DataFile createOneParquetFile(int fileIndex) throws IOException {
+ OutputFile outputFile =
+ table()
+ .io()
+ .newOutputFile(
+ table()
+ .locationProvider()
+ .newDataLocation(String.format(Locale.ROOT, "data-%05d.parquet", fileIndex)));
+
+ // forTable(table()) propagates the table properties (including
+ // PARQUET_ROW_GROUP_SIZE_BYTES set in initTable()) into the writer. Without this,
+ // the writer falls back to the Parquet default row-group size and emits a single
+ // row group for the whole file, defeating row-group skipping.
+ DataWriter<Record> writer =
+ Parquet.writeData(outputFile)
+ .forTable(table())
+ .createWriterFunc(GenericParquetWriter::create)
+ .variantShreddingFunc(shredder)
+ .overwrite()
+ .build();
+
+ writeOneFile(writer, variantMetadata, fileIndex);
+ DataFile dataFile = writer.toDataFile();
+ LOG.info(
+ "File {}: {} bytes, ~{} row groups",
+ fileIndex,
+ dataFile.fileSizeInBytes(),
+ Math.max(1, dataFile.fileSizeInBytes() / ROW_GROUP_SIZE));
+ dumpRowGroupDiagnostics(outputFile);
+ return dataFile;
+ }
+
+ /**
+ * Per-row-group statistics dump. Confirms that:
+ *
+ * <ul>
+ * <li>multiple row groups were actually written (not just one)
+ * <li>category and shredded {@code varcategory} columns have populated min/max stats
+ * <li>clustering produced tight ranges per row group (most groups have {@code min == max})
+ * <li>a filter like {@code varcategory = 1} can theoretically skip groups
+ * </ul>
+ *
+ * <p>Only the shredded path looks at {@code nested.typed_value.varcategory.typed_value}; on
+ * unshredded runs that path doesn't exist and is silently absent in the per-block dump.
+ */
+ private void dumpRowGroupDiagnostics(OutputFile outputFile) throws IOException {
+ ColumnPath categoryPath = ColumnPath.get("category");
+ ColumnPath shreddedVarcategoryPath =
+ ColumnPath.get("nested", "typed_value", "varcategory", "typed_value");
+ int expectedSkipForVarcatEq1 = 0;
+ int blocksMissingShreddedVarcategory = 0;
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(ParquetIO.file(outputFile.toInputFile()))) {
+ List<BlockMetaData> blocks = reader.getFooter().getBlocks();
+ LOG.info("Parquet diagnostics: {} row groups", blocks.size());
+ for (int i = 0; i < blocks.size(); i++) {
+ BlockMetaData block = blocks.get(i);
+ ColumnChunkMetaData categoryChunk = findChunk(block, categoryPath);
+ ColumnChunkMetaData varcategoryChunk = findChunk(block, shreddedVarcategoryPath);
+ String categorySummary = summarizeStats(categoryChunk);
+ String varcategorySummary =
+ varcategoryChunk == null ? "<absent>" : summarizeStats(varcategoryChunk);
+ LOG.info(
+ " block[{}]: rows={} bytes={} category={} shreddedVarcategory={}",
+ i,
+ block.getRowCount(),
+ block.getTotalByteSize(),
+ categorySummary,
+ varcategorySummary);
+ if (isShredded()) {
+ if (varcategoryChunk == null) {
+ blocksMissingShreddedVarcategory++;
+ } else if (rangeExcludesIntLiteral(varcategoryChunk.getStatistics(), 1)) {
+ expectedSkipForVarcatEq1++;
+ }
+ }
+ }
+ if (isShredded()) {
+ if (blocksMissingShreddedVarcategory > 0) {
+ LOG.warn(
+ "Shredded run is missing the varcategory typed_value column in {} of {} blocks "
+ + "- shredding did not produce the expected layout",
+ blocksMissingShreddedVarcategory,
+ blocks.size());
+ }
+ LOG.info(
+ "Parquet diagnostics summary: rowGroups={} expectedSkipForVarcatEq1={}",
+ blocks.size(),
+ expectedSkipForVarcatEq1);
+ }
+ }
+ }
+
+ /**
+ * Find the column chunk for a given column path
+ *
+ * @param block metadata
+ * @param path column path
+ * @return chunk metadata or null if not found.
+ */
+ private static ColumnChunkMetaData findChunk(BlockMetaData block, ColumnPath path) {
+ return block.getColumns().stream()
+ .filter(chunk -> chunk.getPath().equals(path))
+ .findFirst()
+ .orElse(null);
+ }
+
+ /**
+ * Generate a simple summary of the column chunk stats, or <absent> if the supplied metadata
+ * is null, <no-stats> if there were no statistics found.
+ *
+ * @param chunk chunk metadata
+ * @return summary.
+ */
+ private static String summarizeStats(ColumnChunkMetaData chunk) {
+ if (chunk == null) {
+ return "<absent>";
+ }
+ Statistics<?> stats = chunk.getStatistics();
+ if (stats == null || stats.isEmpty()) {
+ return "<no-stats>";
+ }
+ return String.format(
+ Locale.ROOT,
+ "min=%s max=%s nulls=%s values=%d",
+ stats.hasNonNullValue() ? stats.genericGetMin() : "<unset>",
+ stats.hasNonNullValue() ? stats.genericGetMax() : "<unset>",
+ stats.isNumNullsSet() ? Long.toString(stats.getNumNulls()) : "<unset>",
+ chunk.getValueCount());
+ }
+
+ /**
+ * True if the chunk's statistics prove the integer literal cannot appear in the group — i.e. the
+ * row group is skippable for an equality predicate on that literal. Returns false if stats are
+ * absent, empty, or span the literal.
+ */
+ private static boolean rangeExcludesIntLiteral(Statistics<?> stats, int literal) {
+ if (stats == null || stats.isEmpty() || !stats.hasNonNullValue()) {
+ return false;
+ }
+ if (!(stats.genericGetMin() instanceof Number minNum)
+ || !(stats.genericGetMax() instanceof Number maxNum)) {
+ return false;
+ }
+ long min = minNum.longValue();
+ long max = maxNum.longValue();
+ return literal < min || literal > max;
+ }
+
+ /**
+ * Build the Avro table.
+ *
+ * @return the file size.
+ * @throws IOException creation failure.
+ */
+ private long appendAvroVariantData() throws IOException {
+ long size = 0;
+ final AppendFiles append = table().newAppend();
+ for (int fileNum = 0; fileNum < NUM_FILES; fileNum++) {
+ final DataFile dataFile = createOneAvroFile(fileNum);
+ append.appendFile(dataFile);
+ size += dataFile.fileSizeInBytes();
+ }
+ append.commit();
+ return size;
+ }
+
+ /**
+ * Create a single Avro file as part of the append operation.
+ *
+ * @param fileIndex file index to use in generation of name and category derivation.
+ * @return the data file.
+ * @throws IOException write failure.
+ */
+ private DataFile createOneAvroFile(int fileIndex) throws IOException {
+ OutputFile outputFile =
+ table()
+ .io()
+ .newOutputFile(
+ table()
+ .locationProvider()
+ .newDataLocation(String.format(Locale.ROOT, "data-%05d.avro", fileIndex)));
+
+ DataWriter<Record> writer =
+ Avro.writeData(outputFile)
+ .forTable(table())
+ .schema(SCHEMA)
+ .createWriterFunc(org.apache.iceberg.data.avro.DataWriter::create)
+ .overwrite()
+ .withSpec(PartitionSpec.unpartitioned())
+ .build();
+
+ writeOneFile(writer, variantMetadata, fileIndex);
+ return writer.toDataFile();
+ }
+
+ /**
+ * Build the contents of a Parquet or Avro file, close the writer when done.
+ *
+ * <p>Rows are written in category-clustered order (all rows for category 0, then category 1,
+ * etc.) so that each Parquet row group covers only a narrow range of category values. This
+ * enables row-group skipping: a filter such as {@code category IN (5)} skips ~95% of row groups
+ * (those whose min/max range excludes 5) across the ~34 row groups in the file.
+ *
+ * @param writer record writer (will be closed when the method returns)
+ * @param metadata variant metadata
+ * @param fileNum file number in sequence (unused; retained for call-site consistency)
+ * @throws IOException write failure
+ */
+ private void writeOneFile(DataWriter<Record> writer, VariantMetadata metadata, int fileNum)
+ throws IOException {
+ try (writer) {
+ // large base value from filenum
+ long id = fileNum * 10_000_000_000L;
+ GenericRecord record = GenericRecord.create(SCHEMA);
+ int rowsPerCategory = NUM_ROWS_PER_FILE / NUM_CATEGORIES;
+ for (int category = 0; category < NUM_CATEGORIES; category++) {
+ for (int j = 0; j < rowsPerCategory; j++) {
+ id++;
+ Variant variant = buildVariant(metadata, id, category, repeatedStrings[category]);
+ record.setField(COL_ID, id);
+ record.setField("category", category);
+ record.setField(COL_NESTED, variant);
+ record.setField(COL_ARR, buildArrayVariant(id, category));
+ writer.write(record);
+ }
+ }
+ }
+ }
+
+ /**
+ * Build the nested variant structure.
+ *
+ * @param metadata variant metadata
+ * @param id row ID
+ * @param category category
+ * @param col4 string for column 4
+ * @return a variant
+ */
+ private static Variant buildVariant(
+ VariantMetadata metadata, long id, int category, String col4) {
+ ShreddedObject obj = Variants.object(metadata);
+ obj.put("idstr", Variants.of("item_" + id));
+ obj.put("varid", Variants.of(id));
+ obj.put("varcategory", Variants.of(category));
+ obj.put("col4", Variants.of(col4));
+ return Variant.of(metadata, obj);
+ }
+
+ /**
+ * Build a 5-element integer array variant.
+ *
+ * <p>Arrays have no field-name metadata, so {@link Variants#emptyMetadata()} is used.
+ *
+ * @param id row ID
+ * @param category category value
+ * @return a variant holding an array
+ */
+ private static Variant buildArrayVariant(long id, int category) {
+ ValueArray arr = Variants.array();
+ arr.add(Variants.of(category));
+ arr.add(Variants.of((int) (id % 20)));
+ arr.add(Variants.of((int) (id % 100)));
+ arr.add(Variants.of((int) (id % 50)));
+ arr.add(Variants.of((int) (id % 1000)));
+ return Variant.of(Variants.emptyMetadata(), arr);
+ }
+
+ /** Log the parquet metrics invocations. */
+ private void logRowGroupFiltering() {}
Review Comment:
ok, I will cut and when pushdown is added, they can be reinstated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "nssalian (via GitHub)" <gi...@apache.org>.
nssalian commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3564859584
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,915 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RuntimeConfig;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant object
+ * <li>fully shredded variant object
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in a very small range; written in clustered order so each Parquet row group should
+ * contain very few entries from different ranges.
+ * nested: variant (object)
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32
+ * .col4: string :- non-unique string per row (same number as category count)
+ * arr: variant (array, always unshredded)
+ * [0]: int32 :- category
+ * [1]: int32 :- id % 20
+ * [2]: int32 :- id % 100
+ * [3]: int32 :- id % 50
+ * [4]: int32 :- id % 1000
+ * </pre>
+ *
+ * <p>Arrays don't shred because they are anonymous.
+ *
+ * <p>All data is written into a single file to keep the Spark task count to 1. Combined with a
+ * {@code local[1]} Spark master, all benchmark iterations execute on a single thread, which
+ * eliminates inter-task scheduling noise and makes measurements reproducible.
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shredded
+ * parquet, irrespective of performance — look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 5)
+@Measurement(iterations = 5)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 60, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ private static final int NUM_FILES = 1;
+ private static final int NUM_ROWS_PER_FILE = 100_000;
+
+ /** Size of a compressed rowgroup. */
+ private static final int ROW_GROUP_SIZE = 1 * 1024 * 1024;
+
+ /** Number of distinct category values. */
+ private static final int NUM_CATEGORIES = 10;
+
+ public static final String COL_ID = "id";
+ private static final String COL_NESTED = "nested";
+ private static final String COL_ARR = "arr";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ /**
+ * Name of the second Spark temp view for the self-join benchmark. Both views read the same
+ * underlying table; registering two aliases avoids any Spark self-join restrictions.
+ */
+ private static final String TEMP_VIEW2 = "variant_table2";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, COL_NESTED, Types.VariantType.get()), /* The variant object. */
+ required(4, COL_ARR, Types.VariantType.get())); /* The variant array. */
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'bigint')";
+
+ /** Equality check. Pulled out to allow experimentation with set membershop vs. equals. */
+ public static final String EQUALITY_CHECK = " = 1";
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category " + EQUALITY_CHECK;
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}. */
+ private static final String FILTER_ON_NESTED_CATEGORY =
+ VARIANT_GET_NESTED_CATEGORY + EQUALITY_CHECK;
+
+ /**
+ * Get element 0 of the array variant: {@value}.
+ *
+ * <p>Note: {@code variant_get} with a JSONPath {@code $[n]} expression accesses array elements.
+ */
+ private static final String VARIANT_GET_ARR_ELT0 = "variant_get(arr, '$[0]', 'int')";
+
+ /** Filter on element 0 of the array variant: {@value}. */
+ private static final String FILTER_ON_ARR_ELT0 = VARIANT_GET_ARR_ELT0 + EQUALITY_CHECK;
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * Use a single-threaded Spark master to eliminate inter-task scheduling noise.
+ *
+ * <p>With one file and {@code local[1]}, every scan benchmark runs as a single serial task.
+ */
+ @Override
+ protected String sparkMaster() {
+ return "local[1]";
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // large split size keeps the single data file as one Spark task.
+ properties.put(TableProperties.SPLIT_SIZE, Integer.toString(512 * 1024 * 1024));
+ // small row group size ensures multiple row groups per file for row-group skipping benchmarks.
+ properties.put(TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES, Integer.toString(ROW_GROUP_SIZE));
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup(Level.Trial)
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+ final RuntimeConfig sc = spark().conf();
+ // one shuffle partition keeps the join benchmark on a single task.
+ sc.set("spark.sql.shuffle.partitions", "1");
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ isShredded()
+ ? (fieldId, fieldName) -> COL_NESTED.equals(fieldName) ? shreddedType : null
+ : (fieldId, fieldName) -> null;
+
+ final int categoryCount = NUM_CATEGORIES;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ tableDataset().createOrReplaceTempView(TEMP_VIEW2);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ logRowGroupFiltering();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /**
+ * Is the table shredded?
+ *
+ * @return true if the table is shredded.
+ */
+ private boolean isShredded() {
+ return tableType.shredded;
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count(Blackhole blackhole) {
+ project(blackhole, "*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_CATEGORY, false);
+ }
+
+ /**
+ * Perform a sql select operation to non-zero count of rows.
+ *
+ * @param blackhole consumer of all data
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @param expectPushdownWhenShredded on a shredded table, should the rowgroup filter report
+ * invocation?
+ * @return the result.
+ */
+ private long select(
+ final Blackhole blackhole,
+ String projection,
+ String where,
+ boolean expectPushdownWhenShredded) {
+ final long result = sql(blackhole, toSql(projection, where));
+ if (expectPushdownWhenShredded) {
+ expectFilteringOfShreddedFields();
+ }
+ return result;
+ }
+
+ /**
+ * Perform a sql select operation where the result is an empty set.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @param expectPushdownWhenShredded on a shredded table, should the rowgroup filter report
+ * invocation?
+ */
+ private void selectEmptyResults(
+ final String projection, final String where, boolean expectPushdownWhenShredded) {
+ String command = toSql(projection, where);
+ resetMetricsCounter();
+ final Dataset<Row> ds = spark().sql(command);
+ assertThat(ds.collectAsList()).describedAs("Result of command %s", command).isEmpty();
+ if (expectPushdownWhenShredded) {
+ expectFilteringOfShreddedFields();
+ }
+ }
+
+ /**
+ * Create a SQL select statement.
+ *
+ * @param projection projection clause
+ * @param where filter, may be null
+ * @return the string for evaluation.
+ */
+ private static String toSql(final String projection, final String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ return text;
+ }
+
+ /**
+ * Issue any Spark SQL command that returns a non-zero count of rows.
+ *
+ * @param blackhole consumer of all data
+ * @param command command to execute
+ * @return count of result.
+ */
+ private long sql(final Blackhole blackhole, String command) {
+ LOG.info("{}", command);
+ resetMetricsCounter();
+ return materializeNonEmpty(blackhole, command, spark().sql(command));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(final Blackhole blackhole, String projection) {
+ return select(blackhole, projection, null, false);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY, false);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID(Blackhole blackhole) {
+ project(blackhole, COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat(Blackhole blackhole) {
+ select(blackhole, "*", FILTER_ON_CATEGORY, false);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat(Blackhole blackhole) {
+ select(blackhole, "*", FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ @Benchmark
+ public void filterVarCatSetMembership(Blackhole blackhole) {
+ select(blackhole, COL_ID, VARIANT_GET_NESTED_CATEGORY + " IN (1)", true);
+ }
+
+ @Benchmark
+ public void filterVarCatSetNotInMembership(Blackhole blackhole) {
+ selectEmptyResults(COL_ID, VARIANT_GET_NESTED_CATEGORY + " IN (100, 400)", false);
+ }
+
+ @Benchmark
+ public void filterVarCatSetNotInRange(Blackhole blackhole) {
+ selectEmptyResults(COL_ID, VARIANT_GET_NESTED_CATEGORY + " < 0", false);
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ /** selecting on 1, but using a pair of ranges to locate. Two scans needed */
+ @Benchmark
+ public void filterVarcatProjectVarIDRanged(Blackhole blackhole) {
+ select(
+ blackhole,
+ VARIANT_GET_NESTED_ID,
+ VARIANT_GET_NESTED_CATEGORY + " > 0 and " + VARIANT_GET_NESTED_CATEGORY + " < 2",
+ true);
+ }
+
+ /**
+ * Project element 0 of the array variant.
+ *
+ * <p>Note: the {@code arr} column is stored unshredded in all table types (array shredding is out
+ * of scope). This benchmark compares unshredded Parquet vs Avro array element access.
+ */
+ @Benchmark
+ public void projectArrayCategory(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_ARR_ELT0);
+ }
+
+ /** Filter on element 0 of the array variant then project on ID. */
+ @Benchmark
+ public void filterArrayCategoryProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_ARR_ELT0, false);
+ }
+
+ /**
+ * Filter on element 0 of the array variant then project element 0.
+ *
+ * <p>Both the filter and projection touch the same array element; this measures the cost of
+ * evaluating {@code variant_get} on an array twice per row (once for filter, once for output).
+ */
+ @Benchmark
+ public void filterArray0ProjectArray0(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_ARR_ELT0, FILTER_ON_ARR_ELT0, false);
+ }
+
+ /** Explode the array variant into one row per element. */
+ @Benchmark
+ public void explodeArraryColumns(Blackhole blackhole) {
+ sql(blackhole, "select t.* from " + TEMP_VIEW + " as table, lateral variant_explode(arr) as t");
+ }
+
+ /**
+ * Self-join the table on {@code arr[4]} (= {@code id % 1000}, range 0-999) against {@code
+ * varcategory} from the nested variant.
+ *
+ * <p>TODO: failing in spark. Needs investigation.
+ */
+ // @Benchmark
+ public void joinArrElt4OnVarcat(Blackhole blackhole) {
Review Comment:
Could we drop the commented-out `joinArrElt4OnVarcat` and open an issue for the failing self-join instead? A linked issue tracks it better than a `// @Benchmark` + TODO.
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,915 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RuntimeConfig;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant object
+ * <li>fully shredded variant object
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in a very small range; written in clustered order so each Parquet row group should
+ * contain very few entries from different ranges.
+ * nested: variant (object)
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32
+ * .col4: string :- non-unique string per row (same number as category count)
+ * arr: variant (array, always unshredded)
+ * [0]: int32 :- category
+ * [1]: int32 :- id % 20
+ * [2]: int32 :- id % 100
+ * [3]: int32 :- id % 50
+ * [4]: int32 :- id % 1000
+ * </pre>
+ *
+ * <p>Arrays don't shred because they are anonymous.
+ *
+ * <p>All data is written into a single file to keep the Spark task count to 1. Combined with a
+ * {@code local[1]} Spark master, all benchmark iterations execute on a single thread, which
+ * eliminates inter-task scheduling noise and makes measurements reproducible.
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shredded
+ * parquet, irrespective of performance — look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 5)
+@Measurement(iterations = 5)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 60, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ private static final int NUM_FILES = 1;
+ private static final int NUM_ROWS_PER_FILE = 100_000;
+
+ /** Size of a compressed rowgroup. */
+ private static final int ROW_GROUP_SIZE = 1 * 1024 * 1024;
+
+ /** Number of distinct category values. */
+ private static final int NUM_CATEGORIES = 10;
+
+ public static final String COL_ID = "id";
Review Comment:
Minor: `COL_ID` and `EQUALITY_CHECK` are `public` while `COL_NESTED`/`COL_ARR` right below are `private`, and the sibling benchmarks keep their constants private. Same for `TableType` (and `Depth`/`DEEP_NEST_DEPTH`/`ITERATIONS` in the core benchmark). Worth making these private to match.
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,915 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RuntimeConfig;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant object
+ * <li>fully shredded variant object
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in a very small range; written in clustered order so each Parquet row group should
+ * contain very few entries from different ranges.
+ * nested: variant (object)
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32
+ * .col4: string :- non-unique string per row (same number as category count)
+ * arr: variant (array, always unshredded)
+ * [0]: int32 :- category
+ * [1]: int32 :- id % 20
+ * [2]: int32 :- id % 100
+ * [3]: int32 :- id % 50
+ * [4]: int32 :- id % 1000
+ * </pre>
+ *
+ * <p>Arrays don't shred because they are anonymous.
+ *
+ * <p>All data is written into a single file to keep the Spark task count to 1. Combined with a
+ * {@code local[1]} Spark master, all benchmark iterations execute on a single thread, which
+ * eliminates inter-task scheduling noise and makes measurements reproducible.
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shredded
+ * parquet, irrespective of performance — look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 5)
+@Measurement(iterations = 5)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 60, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ private static final int NUM_FILES = 1;
+ private static final int NUM_ROWS_PER_FILE = 100_000;
+
+ /** Size of a compressed rowgroup. */
+ private static final int ROW_GROUP_SIZE = 1 * 1024 * 1024;
+
+ /** Number of distinct category values. */
+ private static final int NUM_CATEGORIES = 10;
+
+ public static final String COL_ID = "id";
+ private static final String COL_NESTED = "nested";
+ private static final String COL_ARR = "arr";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ /**
+ * Name of the second Spark temp view for the self-join benchmark. Both views read the same
+ * underlying table; registering two aliases avoids any Spark self-join restrictions.
+ */
+ private static final String TEMP_VIEW2 = "variant_table2";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, COL_NESTED, Types.VariantType.get()), /* The variant object. */
+ required(4, COL_ARR, Types.VariantType.get())); /* The variant array. */
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'bigint')";
+
+ /** Equality check. Pulled out to allow experimentation with set membershop vs. equals. */
+ public static final String EQUALITY_CHECK = " = 1";
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category " + EQUALITY_CHECK;
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}. */
+ private static final String FILTER_ON_NESTED_CATEGORY =
+ VARIANT_GET_NESTED_CATEGORY + EQUALITY_CHECK;
+
+ /**
+ * Get element 0 of the array variant: {@value}.
+ *
+ * <p>Note: {@code variant_get} with a JSONPath {@code $[n]} expression accesses array elements.
+ */
+ private static final String VARIANT_GET_ARR_ELT0 = "variant_get(arr, '$[0]', 'int')";
+
+ /** Filter on element 0 of the array variant: {@value}. */
+ private static final String FILTER_ON_ARR_ELT0 = VARIANT_GET_ARR_ELT0 + EQUALITY_CHECK;
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * Use a single-threaded Spark master to eliminate inter-task scheduling noise.
+ *
+ * <p>With one file and {@code local[1]}, every scan benchmark runs as a single serial task.
+ */
+ @Override
+ protected String sparkMaster() {
+ return "local[1]";
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // large split size keeps the single data file as one Spark task.
+ properties.put(TableProperties.SPLIT_SIZE, Integer.toString(512 * 1024 * 1024));
+ // small row group size ensures multiple row groups per file for row-group skipping benchmarks.
+ properties.put(TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES, Integer.toString(ROW_GROUP_SIZE));
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup(Level.Trial)
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+ final RuntimeConfig sc = spark().conf();
+ // one shuffle partition keeps the join benchmark on a single task.
+ sc.set("spark.sql.shuffle.partitions", "1");
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ isShredded()
+ ? (fieldId, fieldName) -> COL_NESTED.equals(fieldName) ? shreddedType : null
+ : (fieldId, fieldName) -> null;
+
+ final int categoryCount = NUM_CATEGORIES;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ tableDataset().createOrReplaceTempView(TEMP_VIEW2);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ logRowGroupFiltering();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /**
+ * Is the table shredded?
+ *
+ * @return true if the table is shredded.
+ */
+ private boolean isShredded() {
+ return tableType.shredded;
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count(Blackhole blackhole) {
+ project(blackhole, "*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_CATEGORY, false);
+ }
+
+ /**
+ * Perform a sql select operation to non-zero count of rows.
+ *
+ * @param blackhole consumer of all data
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @param expectPushdownWhenShredded on a shredded table, should the rowgroup filter report
+ * invocation?
+ * @return the result.
+ */
+ private long select(
+ final Blackhole blackhole,
+ String projection,
+ String where,
+ boolean expectPushdownWhenShredded) {
+ final long result = sql(blackhole, toSql(projection, where));
+ if (expectPushdownWhenShredded) {
+ expectFilteringOfShreddedFields();
+ }
+ return result;
+ }
+
+ /**
+ * Perform a sql select operation where the result is an empty set.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @param expectPushdownWhenShredded on a shredded table, should the rowgroup filter report
+ * invocation?
+ */
+ private void selectEmptyResults(
+ final String projection, final String where, boolean expectPushdownWhenShredded) {
+ String command = toSql(projection, where);
+ resetMetricsCounter();
+ final Dataset<Row> ds = spark().sql(command);
+ assertThat(ds.collectAsList()).describedAs("Result of command %s", command).isEmpty();
+ if (expectPushdownWhenShredded) {
+ expectFilteringOfShreddedFields();
+ }
+ }
+
+ /**
+ * Create a SQL select statement.
+ *
+ * @param projection projection clause
+ * @param where filter, may be null
+ * @return the string for evaluation.
+ */
+ private static String toSql(final String projection, final String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ return text;
+ }
+
+ /**
+ * Issue any Spark SQL command that returns a non-zero count of rows.
+ *
+ * @param blackhole consumer of all data
+ * @param command command to execute
+ * @return count of result.
+ */
+ private long sql(final Blackhole blackhole, String command) {
+ LOG.info("{}", command);
+ resetMetricsCounter();
+ return materializeNonEmpty(blackhole, command, spark().sql(command));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(final Blackhole blackhole, String projection) {
+ return select(blackhole, projection, null, false);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY, false);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID(Blackhole blackhole) {
+ project(blackhole, COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat(Blackhole blackhole) {
+ select(blackhole, "*", FILTER_ON_CATEGORY, false);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat(Blackhole blackhole) {
+ select(blackhole, "*", FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ @Benchmark
+ public void filterVarCatSetMembership(Blackhole blackhole) {
+ select(blackhole, COL_ID, VARIANT_GET_NESTED_CATEGORY + " IN (1)", true);
+ }
+
+ @Benchmark
+ public void filterVarCatSetNotInMembership(Blackhole blackhole) {
+ selectEmptyResults(COL_ID, VARIANT_GET_NESTED_CATEGORY + " IN (100, 400)", false);
+ }
+
+ @Benchmark
+ public void filterVarCatSetNotInRange(Blackhole blackhole) {
+ selectEmptyResults(COL_ID, VARIANT_GET_NESTED_CATEGORY + " < 0", false);
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY, true);
+ }
+
+ /** selecting on 1, but using a pair of ranges to locate. Two scans needed */
+ @Benchmark
+ public void filterVarcatProjectVarIDRanged(Blackhole blackhole) {
+ select(
+ blackhole,
+ VARIANT_GET_NESTED_ID,
+ VARIANT_GET_NESTED_CATEGORY + " > 0 and " + VARIANT_GET_NESTED_CATEGORY + " < 2",
+ true);
+ }
+
+ /**
+ * Project element 0 of the array variant.
+ *
+ * <p>Note: the {@code arr} column is stored unshredded in all table types (array shredding is out
+ * of scope). This benchmark compares unshredded Parquet vs Avro array element access.
+ */
+ @Benchmark
+ public void projectArrayCategory(Blackhole blackhole) {
+ project(blackhole, VARIANT_GET_ARR_ELT0);
+ }
+
+ /** Filter on element 0 of the array variant then project on ID. */
+ @Benchmark
+ public void filterArrayCategoryProjectID(Blackhole blackhole) {
+ select(blackhole, COL_ID, FILTER_ON_ARR_ELT0, false);
+ }
+
+ /**
+ * Filter on element 0 of the array variant then project element 0.
+ *
+ * <p>Both the filter and projection touch the same array element; this measures the cost of
+ * evaluating {@code variant_get} on an array twice per row (once for filter, once for output).
+ */
+ @Benchmark
+ public void filterArray0ProjectArray0(Blackhole blackhole) {
+ select(blackhole, VARIANT_GET_ARR_ELT0, FILTER_ON_ARR_ELT0, false);
+ }
+
+ /** Explode the array variant into one row per element. */
+ @Benchmark
+ public void explodeArraryColumns(Blackhole blackhole) {
+ sql(blackhole, "select t.* from " + TEMP_VIEW + " as table, lateral variant_explode(arr) as t");
+ }
+
+ /**
+ * Self-join the table on {@code arr[4]} (= {@code id % 1000}, range 0-999) against {@code
+ * varcategory} from the nested variant.
+ *
+ * <p>TODO: failing in spark. Needs investigation.
+ */
+ // @Benchmark
+ public void joinArrElt4OnVarcat(Blackhole blackhole) {
+ final String text =
+ "SELECT t1."
+ + COL_ID
+ + ", t2."
+ + COL_ID
+ + " FROM "
+ + TEMP_VIEW
+ + " t1 JOIN "
+ + TEMP_VIEW2
+ + " t2 ON "
+ + "variant_get(t1.arr, '$[4]', 'int') = variant_get(t2.nested, '$.varcategory', 'int')";
+ sql(blackhole, text);
+ expectFilteringOfShreddedFields();
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param blackhole consumer of all data
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(final Blackhole blackhole, String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.queryExecution().toRdd().toJavaRDD().count();
+ blackhole.consume(count);
+ Preconditions.checkState(
+ count > 0, "Operation %s on %s table returned no results", operation, tableType);
+ LOG.info("Materialization of Dataset returned {} records", count);
+ return count;
+ }
+
+ /**
+ * Generate files in the target subdir, sorted by timestamp and with their format included in the
+ * path.
+ *
+ * @return the location for the new table
+ */
+ @Override
+ protected String newTableLocation() {
+ Path tablePath =
+ new Path(
+ String.format(
+ Locale.ROOT,
+ "build/tables/spark-iceberg-table-%d-%s",
+ System.currentTimeMillis(),
+ UUID.randomUUID()));
+ return tablePath.toString();
+ }
+
+ /**
+ * Build the parquet dataset.
+ *
+ * @throws IOException creation failure.
+ */
+ private long appendVariantData() throws IOException {
+ long size = 0;
+ final AppendFiles append = table().newAppend();
+
+ for (int fileNum = 0; fileNum < NUM_FILES; fileNum++) {
+ final DataFile dataFile = createOneParquetFile(fileNum);
+ append.appendFile(dataFile);
+ size += dataFile.fileSizeInBytes();
+ }
+ append.commit();
+ return size;
+ }
+
+ /**
+ * Create a single parquet file as part of the append operation.
+ *
+ * @param fileIndex file index to use in generation of name and category derivation.
+ * @return the data file.
+ * @throws IOException write failure.
+ */
+ private DataFile createOneParquetFile(int fileIndex) throws IOException {
+ OutputFile outputFile =
+ table()
+ .io()
+ .newOutputFile(
+ table()
+ .locationProvider()
+ .newDataLocation(String.format(Locale.ROOT, "data-%05d.parquet", fileIndex)));
+
+ // forTable(table()) propagates the table properties (including
+ // PARQUET_ROW_GROUP_SIZE_BYTES set in initTable()) into the writer. Without this,
+ // the writer falls back to the Parquet default row-group size and emits a single
+ // row group for the whole file, defeating row-group skipping.
+ DataWriter<Record> writer =
+ Parquet.writeData(outputFile)
+ .forTable(table())
+ .createWriterFunc(GenericParquetWriter::create)
+ .variantShreddingFunc(shredder)
+ .overwrite()
+ .build();
+
+ writeOneFile(writer, variantMetadata, fileIndex);
+ DataFile dataFile = writer.toDataFile();
+ LOG.info(
+ "File {}: {} bytes, ~{} row groups",
+ fileIndex,
+ dataFile.fileSizeInBytes(),
+ Math.max(1, dataFile.fileSizeInBytes() / ROW_GROUP_SIZE));
+ dumpRowGroupDiagnostics(outputFile);
+ return dataFile;
+ }
+
+ /**
+ * Per-row-group statistics dump. Confirms that:
+ *
+ * <ul>
+ * <li>multiple row groups were actually written (not just one)
+ * <li>category and shredded {@code varcategory} columns have populated min/max stats
+ * <li>clustering produced tight ranges per row group (most groups have {@code min == max})
+ * <li>a filter like {@code varcategory = 1} can theoretically skip groups
+ * </ul>
+ *
+ * <p>Only the shredded path looks at {@code nested.typed_value.varcategory.typed_value}; on
+ * unshredded runs that path doesn't exist and is silently absent in the per-block dump.
+ */
+ private void dumpRowGroupDiagnostics(OutputFile outputFile) throws IOException {
+ ColumnPath categoryPath = ColumnPath.get("category");
+ ColumnPath shreddedVarcategoryPath =
+ ColumnPath.get("nested", "typed_value", "varcategory", "typed_value");
+ int expectedSkipForVarcatEq1 = 0;
+ int blocksMissingShreddedVarcategory = 0;
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(ParquetIO.file(outputFile.toInputFile()))) {
+ List<BlockMetaData> blocks = reader.getFooter().getBlocks();
+ LOG.info("Parquet diagnostics: {} row groups", blocks.size());
+ for (int i = 0; i < blocks.size(); i++) {
+ BlockMetaData block = blocks.get(i);
+ ColumnChunkMetaData categoryChunk = findChunk(block, categoryPath);
+ ColumnChunkMetaData varcategoryChunk = findChunk(block, shreddedVarcategoryPath);
+ String categorySummary = summarizeStats(categoryChunk);
+ String varcategorySummary =
+ varcategoryChunk == null ? "<absent>" : summarizeStats(varcategoryChunk);
+ LOG.info(
+ " block[{}]: rows={} bytes={} category={} shreddedVarcategory={}",
+ i,
+ block.getRowCount(),
+ block.getTotalByteSize(),
+ categorySummary,
+ varcategorySummary);
+ if (isShredded()) {
+ if (varcategoryChunk == null) {
+ blocksMissingShreddedVarcategory++;
+ } else if (rangeExcludesIntLiteral(varcategoryChunk.getStatistics(), 1)) {
+ expectedSkipForVarcatEq1++;
+ }
+ }
+ }
+ if (isShredded()) {
+ if (blocksMissingShreddedVarcategory > 0) {
+ LOG.warn(
+ "Shredded run is missing the varcategory typed_value column in {} of {} blocks "
+ + "- shredding did not produce the expected layout",
+ blocksMissingShreddedVarcategory,
+ blocks.size());
+ }
+ LOG.info(
+ "Parquet diagnostics summary: rowGroups={} expectedSkipForVarcatEq1={}",
+ blocks.size(),
+ expectedSkipForVarcatEq1);
+ }
+ }
+ }
+
+ /**
+ * Find the column chunk for a given column path
+ *
+ * @param block metadata
+ * @param path column path
+ * @return chunk metadata or null if not found.
+ */
+ private static ColumnChunkMetaData findChunk(BlockMetaData block, ColumnPath path) {
+ return block.getColumns().stream()
+ .filter(chunk -> chunk.getPath().equals(path))
+ .findFirst()
+ .orElse(null);
+ }
+
+ /**
+ * Generate a simple summary of the column chunk stats, or <absent> if the supplied metadata
+ * is null, <no-stats> if there were no statistics found.
+ *
+ * @param chunk chunk metadata
+ * @return summary.
+ */
+ private static String summarizeStats(ColumnChunkMetaData chunk) {
+ if (chunk == null) {
+ return "<absent>";
+ }
+ Statistics<?> stats = chunk.getStatistics();
+ if (stats == null || stats.isEmpty()) {
+ return "<no-stats>";
+ }
+ return String.format(
+ Locale.ROOT,
+ "min=%s max=%s nulls=%s values=%d",
+ stats.hasNonNullValue() ? stats.genericGetMin() : "<unset>",
+ stats.hasNonNullValue() ? stats.genericGetMax() : "<unset>",
+ stats.isNumNullsSet() ? Long.toString(stats.getNumNulls()) : "<unset>",
+ chunk.getValueCount());
+ }
+
+ /**
+ * True if the chunk's statistics prove the integer literal cannot appear in the group — i.e. the
+ * row group is skippable for an equality predicate on that literal. Returns false if stats are
+ * absent, empty, or span the literal.
+ */
+ private static boolean rangeExcludesIntLiteral(Statistics<?> stats, int literal) {
+ if (stats == null || stats.isEmpty() || !stats.hasNonNullValue()) {
+ return false;
+ }
+ if (!(stats.genericGetMin() instanceof Number minNum)
+ || !(stats.genericGetMax() instanceof Number maxNum)) {
+ return false;
+ }
+ long min = minNum.longValue();
+ long max = maxNum.longValue();
+ return literal < min || literal > max;
+ }
+
+ /**
+ * Build the Avro table.
+ *
+ * @return the file size.
+ * @throws IOException creation failure.
+ */
+ private long appendAvroVariantData() throws IOException {
+ long size = 0;
+ final AppendFiles append = table().newAppend();
+ for (int fileNum = 0; fileNum < NUM_FILES; fileNum++) {
+ final DataFile dataFile = createOneAvroFile(fileNum);
+ append.appendFile(dataFile);
+ size += dataFile.fileSizeInBytes();
+ }
+ append.commit();
+ return size;
+ }
+
+ /**
+ * Create a single Avro file as part of the append operation.
+ *
+ * @param fileIndex file index to use in generation of name and category derivation.
+ * @return the data file.
+ * @throws IOException write failure.
+ */
+ private DataFile createOneAvroFile(int fileIndex) throws IOException {
+ OutputFile outputFile =
+ table()
+ .io()
+ .newOutputFile(
+ table()
+ .locationProvider()
+ .newDataLocation(String.format(Locale.ROOT, "data-%05d.avro", fileIndex)));
+
+ DataWriter<Record> writer =
+ Avro.writeData(outputFile)
+ .forTable(table())
+ .schema(SCHEMA)
+ .createWriterFunc(org.apache.iceberg.data.avro.DataWriter::create)
+ .overwrite()
+ .withSpec(PartitionSpec.unpartitioned())
+ .build();
+
+ writeOneFile(writer, variantMetadata, fileIndex);
+ return writer.toDataFile();
+ }
+
+ /**
+ * Build the contents of a Parquet or Avro file, close the writer when done.
+ *
+ * <p>Rows are written in category-clustered order (all rows for category 0, then category 1,
+ * etc.) so that each Parquet row group covers only a narrow range of category values. This
+ * enables row-group skipping: a filter such as {@code category IN (5)} skips ~95% of row groups
+ * (those whose min/max range excludes 5) across the ~34 row groups in the file.
+ *
+ * @param writer record writer (will be closed when the method returns)
+ * @param metadata variant metadata
+ * @param fileNum file number in sequence (unused; retained for call-site consistency)
+ * @throws IOException write failure
+ */
+ private void writeOneFile(DataWriter<Record> writer, VariantMetadata metadata, int fileNum)
+ throws IOException {
+ try (writer) {
+ // large base value from filenum
+ long id = fileNum * 10_000_000_000L;
+ GenericRecord record = GenericRecord.create(SCHEMA);
+ int rowsPerCategory = NUM_ROWS_PER_FILE / NUM_CATEGORIES;
+ for (int category = 0; category < NUM_CATEGORIES; category++) {
+ for (int j = 0; j < rowsPerCategory; j++) {
+ id++;
+ Variant variant = buildVariant(metadata, id, category, repeatedStrings[category]);
+ record.setField(COL_ID, id);
+ record.setField("category", category);
+ record.setField(COL_NESTED, variant);
+ record.setField(COL_ARR, buildArrayVariant(id, category));
+ writer.write(record);
+ }
+ }
+ }
+ }
+
+ /**
+ * Build the nested variant structure.
+ *
+ * @param metadata variant metadata
+ * @param id row ID
+ * @param category category
+ * @param col4 string for column 4
+ * @return a variant
+ */
+ private static Variant buildVariant(
+ VariantMetadata metadata, long id, int category, String col4) {
+ ShreddedObject obj = Variants.object(metadata);
+ obj.put("idstr", Variants.of("item_" + id));
+ obj.put("varid", Variants.of(id));
+ obj.put("varcategory", Variants.of(category));
+ obj.put("col4", Variants.of(col4));
+ return Variant.of(metadata, obj);
+ }
+
+ /**
+ * Build a 5-element integer array variant.
+ *
+ * <p>Arrays have no field-name metadata, so {@link Variants#emptyMetadata()} is used.
+ *
+ * @param id row ID
+ * @param category category value
+ * @return a variant holding an array
+ */
+ private static Variant buildArrayVariant(long id, int category) {
+ ValueArray arr = Variants.array();
+ arr.add(Variants.of(category));
+ arr.add(Variants.of((int) (id % 20)));
+ arr.add(Variants.of((int) (id % 100)));
+ arr.add(Variants.of((int) (id % 50)));
+ arr.add(Variants.of((int) (id % 1000)));
+ return Variant.of(Variants.emptyMetadata(), arr);
+ }
+
+ /** Log the parquet metrics invocations. */
+ private void logRowGroupFiltering() {}
Review Comment:
`logRowGroupFiltering()`, `resetMetricsCounter()`, and `expectFilteringOfShreddedFields()` are all empty here but their javadoc describes behavior that doesn't exist, and `expectPushdownWhenShredded` is plumbed through `select`/`selectEmptyResults` just to gate the no-op `expectFilteringOfShreddedFields()`. Could these either be implemented or removed (along with the `expectPushdownWhenShredded` param)? Leaving them empty is a bit misleading to the next reader.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3166874024
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}.. */
+ private static final String FILTER_ON_NESTED_CATEGORY = VARIANT_GET_NESTED_CATEGORY + " = 5";
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // turn off compression to remove it as a factor.
+ properties.put(TableProperties.METADATA_COMPRESSION, "none");
+ properties.put(TableProperties.PARQUET_COMPRESSION, "none");
+ properties.put(TableProperties.AVRO_COMPRESSION, "none");
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ tableType.shredded ? (fieldId, fieldName) -> shreddedType : (fieldId, fieldName) -> null;
+
+ // algorithm for category produces a value 0..19
+ final int categoryCount = 20;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count() {
+ project("*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID() {
+ select(COL_ID, FILTER_ON_CATEGORY);
+ }
+
+ /**
+ * Perform a sql select operation.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @return the result.
+ */
+ private long select(String projection, String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ LOG.info("{}", text);
+ return materializeNonEmpty(text, spark().sql(text));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(String projection) {
+ return select(projection, null);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID() {
+ project(COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID() {
+ project(VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID() {
+ select(COL_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat() {
+ select("*", FILTER_ON_CATEGORY);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat() {
+ select("*", FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Project on the category column. */
+ @Benchmark
+ public void projectCat() {
+ project("category");
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat() {
+ project(VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.count();
Review Comment:
it seemed to work, but I will get and discard each record instead
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "rashworld-max (via GitHub)" <gi...@apache.org>.
rashworld-max commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3172900941
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}.. */
+ private static final String FILTER_ON_NESTED_CATEGORY = VARIANT_GET_NESTED_CATEGORY + " = 5";
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // turn off compression to remove it as a factor.
+ properties.put(TableProperties.METADATA_COMPRESSION, "none");
+ properties.put(TableProperties.PARQUET_COMPRESSION, "none");
+ properties.put(TableProperties.AVRO_COMPRESSION, "none");
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ tableType.shredded ? (fieldId, fieldName) -> shreddedType : (fieldId, fieldName) -> null;
+
+ // algorithm for category produces a value 0..19
+ final int categoryCount = 20;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count() {
+ project("*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID() {
+ select(COL_ID, FILTER_ON_CATEGORY);
+ }
+
+ /**
+ * Perform a sql select operation.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @return the result.
+ */
+ private long select(String projection, String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ LOG.info("{}", text);
+ return materializeNonEmpty(text, spark().sql(text));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(String projection) {
+ return select(projection, null);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID() {
+ project(COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID() {
+ project(VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID() {
+ select(COL_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat() {
+ select("*", FILTER_ON_CATEGORY);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat() {
+ select("*", FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Project on the category column. */
+ @Benchmark
+ public void projectCat() {
+ project("category");
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat() {
+ project(VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.count();
Review Comment:
Rashworld-max
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3317822834
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
Review Comment:
moved to `bigint`. Not doing enough with that type to measure any difference between that and int processing in the workflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4788105874
still ready for review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-5374268635
@manuzhang thanks for marking as not stale. I am now ex cloudera, laptopless and mostly restricting my coding to learning golang for the fun of it right now. Maybe I'll get set up for java dev next week.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4408363767
switching to draft again as I'm reworking the benchmark to show variant rowgroup filtering of shredded variants works (#15510), with changes including
- move to single thread spark worker (less variance in results)
- going to a single large file with multiple rowgroups
- exploring the difference between `variant_get(...) is 5` and `variant_get(...) in (5)` to see if spark is treating them differently.
Needs an extra PR in iceberg from qlong and a snaspshot of spark 4.1 with his changes for spark to pass variant_get down
So, surprisingly complex. If I can show the chain works then it's time to start with the feature merges in spark and then here.
This branch will merge without direct dependency, it's just a key goal of the spark benchmark is "show pushdown working". It's not ready to merge unless it can do that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3588099705
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,915 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RuntimeConfig;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant object
+ * <li>fully shredded variant object
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in a very small range; written in clustered order so each Parquet row group should
+ * contain very few entries from different ranges.
+ * nested: variant (object)
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32
+ * .col4: string :- non-unique string per row (same number as category count)
+ * arr: variant (array, always unshredded)
+ * [0]: int32 :- category
+ * [1]: int32 :- id % 20
+ * [2]: int32 :- id % 100
+ * [3]: int32 :- id % 50
+ * [4]: int32 :- id % 1000
+ * </pre>
+ *
+ * <p>Arrays don't shred because they are anonymous.
+ *
+ * <p>All data is written into a single file to keep the Spark task count to 1. Combined with a
+ * {@code local[1]} Spark master, all benchmark iterations execute on a single thread, which
+ * eliminates inter-task scheduling noise and makes measurements reproducible.
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shredded
+ * parquet, irrespective of performance — look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 5)
+@Measurement(iterations = 5)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 60, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ private static final int NUM_FILES = 1;
+ private static final int NUM_ROWS_PER_FILE = 100_000;
+
+ /** Size of a compressed rowgroup. */
+ private static final int ROW_GROUP_SIZE = 1 * 1024 * 1024;
+
+ /** Number of distinct category values. */
+ private static final int NUM_CATEGORIES = 10;
+
+ public static final String COL_ID = "id";
Review Comment:
I'll make them all private,inevitably just ide refactoring setup
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3148157906
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}.. */
+ private static final String FILTER_ON_NESTED_CATEGORY = VARIANT_GET_NESTED_CATEGORY + " = 5";
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // turn off compression to remove it as a factor.
+ properties.put(TableProperties.METADATA_COMPRESSION, "none");
+ properties.put(TableProperties.PARQUET_COMPRESSION, "none");
+ properties.put(TableProperties.AVRO_COMPRESSION, "none");
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ tableType.shredded ? (fieldId, fieldName) -> shreddedType : (fieldId, fieldName) -> null;
+
+ // algorithm for category produces a value 0..19
+ final int categoryCount = 20;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count() {
+ project("*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID() {
+ select(COL_ID, FILTER_ON_CATEGORY);
+ }
+
+ /**
+ * Perform a sql select operation.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @return the result.
+ */
+ private long select(String projection, String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ LOG.info("{}", text);
+ return materializeNonEmpty(text, spark().sql(text));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(String projection) {
+ return select(projection, null);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID() {
+ project(COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID() {
+ project(VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID() {
+ select(COL_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat() {
+ select("*", FILTER_ON_CATEGORY);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat() {
+ select("*", FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Project on the category column. */
+ @Benchmark
+ public void projectCat() {
+ project("category");
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat() {
+ project(VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.count();
Review Comment:
I needed something to do the entire compute and count() worked. Otherwise it's evaluate every row and feed to a black hole. What do you prefer?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3070168785
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java:
##########
@@ -57,7 +58,8 @@
import org.apache.parquet.schema.Type;
import org.apache.parquet.schema.Types;
-class ParquetVariantUtil {
+@VisibleForTesting
+public final class ParquetVariantUtil {
Review Comment:
happy to do that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4246777690
I think this is ready for review. I've got the initial results and it's good for PRs like #3477 to be able to before/after benchmarks.
More stuff can go in later; I've outlined them in my report. Equality deletes would be a fun one
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "rashworld-max (via GitHub)" <gi...@apache.org>.
rashworld-max commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3070233204
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java:
##########
@@ -57,7 +58,8 @@
import org.apache.parquet.schema.Type;
import org.apache.parquet.schema.Types;
-class ParquetVariantUtil {
+@VisibleForTesting
+public final class ParquetVariantUtil {
Review Comment:
Hi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3094427772
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}.. */
+ private static final String FILTER_ON_NESTED_CATEGORY = VARIANT_GET_NESTED_CATEGORY + " = 5";
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // turn off compression to remove it as a factor.
+ properties.put(TableProperties.METADATA_COMPRESSION, "none");
+ properties.put(TableProperties.PARQUET_COMPRESSION, "none");
+ properties.put(TableProperties.AVRO_COMPRESSION, "none");
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ tableType.shredded ? (fieldId, fieldName) -> shreddedType : (fieldId, fieldName) -> null;
+
+ // algorithm for category produces a value 0..19
+ final int categoryCount = 20;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count() {
+ project("*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID() {
+ select(COL_ID, FILTER_ON_CATEGORY);
+ }
+
+ /**
+ * Perform a sql select operation.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @return the result.
+ */
+ private long select(String projection, String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ LOG.info("{}", text);
+ return materializeNonEmpty(text, spark().sql(text));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(String projection) {
+ return select(projection, null);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID() {
+ project(COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID() {
+ project(VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID() {
+ select(COL_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat() {
+ select("*", FILTER_ON_CATEGORY);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat() {
+ select("*", FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Project on the category column. */
+ @Benchmark
+ public void projectCat() {
+ project("category");
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat() {
+ project(VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.count();
Review Comment:
Spark doesn't need to evaluate projection to count records.
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
Review Comment:
should this be `int64` as in the comments above?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-5310407348
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4624673394
@rdblue this pr is ready for review in that "the data it generates, the layout and the queries are sufficient to show tangible speedup with qlong's file skipping and my predicate pushdown changes"
If you have suggestions, I can get back to them from June 22.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4811188792
@Fokko this is ready for its final look-through; i've rebased it but only the final two commits are new, one for the javadocs and the final one with real code changes
@nssalian what do you think here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3166875900
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
Review Comment:
will review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3288022996
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}.. */
+ private static final String FILTER_ON_NESTED_CATEGORY = VARIANT_GET_NESTED_CATEGORY + " = 5";
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // turn off compression to remove it as a factor.
+ properties.put(TableProperties.METADATA_COMPRESSION, "none");
+ properties.put(TableProperties.PARQUET_COMPRESSION, "none");
+ properties.put(TableProperties.AVRO_COMPRESSION, "none");
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ tableType.shredded ? (fieldId, fieldName) -> shreddedType : (fieldId, fieldName) -> null;
+
+ // algorithm for category produces a value 0..19
+ final int categoryCount = 20;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count() {
+ project("*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID() {
+ select(COL_ID, FILTER_ON_CATEGORY);
+ }
+
+ /**
+ * Perform a sql select operation.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @return the result.
+ */
+ private long select(String projection, String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ LOG.info("{}", text);
+ return materializeNonEmpty(text, spark().sql(text));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(String projection) {
+ return select(projection, null);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID() {
+ project(COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID() {
+ project(VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID() {
+ select(COL_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat() {
+ select("*", FILTER_ON_CATEGORY);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat() {
+ select("*", FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Project on the category column. */
+ @Benchmark
+ public void projectCat() {
+ project("category");
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat() {
+ project(VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.count();
Review Comment:
@manuzhang fyi, now using the same sequence as the IcebergSourceBenchmark superclass, with the retention of the count for use in assertions
```
final long count = ds.queryExecution().toRdd().toJavaRDD().count();
blackhole.consume(count);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4168993932
@rashworld-max still a WiP I'm afraid. Need to know I'm measuring the right thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3065715773
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java:
##########
@@ -57,7 +58,8 @@
import org.apache.parquet.schema.Type;
import org.apache.parquet.schema.Types;
-class ParquetVariantUtil {
+@VisibleForTesting
+public final class ParquetVariantUtil {
Review Comment:
Is it possible to relocate the tests rather than expose this? We _can_ do this, but generally prefer not to if we can avoid it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4566985932
@nssalian what do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "nssalian (via GitHub)" <gi...@apache.org>.
nssalian commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-5387986836
@steveloughran it might be hard to get this in this release but I can try doing this after if you don't get a chance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran closed pull request #15629: Core: Add JMH benchmarks for Variants
URL: https://github.com/apache/iceberg/pull/15629
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3070168785
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java:
##########
@@ -57,7 +58,8 @@
import org.apache.parquet.schema.Type;
import org.apache.parquet.schema.Types;
-class ParquetVariantUtil {
+@VisibleForTesting
+public final class ParquetVariantUtil {
Review Comment:
happy to do that...I have done it in the parquet PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "rashworld-max (via GitHub)" <gi...@apache.org>.
rashworld-max commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3070233511
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java:
##########
@@ -57,7 +58,8 @@
import org.apache.parquet.schema.Type;
import org.apache.parquet.schema.Types;
-class ParquetVariantUtil {
+@VisibleForTesting
+public final class ParquetVariantUtil {
Review Comment:
Ho
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#discussion_r3148315078
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/parquet/IcebergSourceVariantIOBenchmark.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.IcebergSourceBenchmark;
+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.Variants;
+import org.apache.parquet.schema.Type;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Timeout;
+import org.openjdk.jmh.annotations.Warmup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A benchmark that evaluates the performance of reading and filtering variant data through
+ * Iceberg's Spark data source.
+ *
+ * <p>Three tables are created with identical data:
+ *
+ * <ol>
+ * <li>avro records containing the bytes
+ * <li>unshredded variant
+ * <li>fully shredded variant
+ * </ol>
+ *
+ * <p>The table structure is:
+ *
+ * <pre>
+ * id: int64 :- unique per row
+ * category: int32 :- in range 0-19: (fileNum % 2) * 10 + (id % 10);
+ * nested: variant
+ * .idstr: string :- unique string per row
+ * .varid: int64 :- id
+ * .varcategory: int32 :- category (0-19)
+ * .col4: string :- non-unique string per row (picked from 20 values based on category)
+ * </pre>
+ *
+ * <p>The structure and repetition helps highlight the space-saving benefits of parquet and shedded
+ * parquet, irrespective of performance -look in the output logs for "Size of Table" to see the
+ * values.
+ *
+ * <p>To run this benchmark for spark-4.1: <code>
+ * ./gradlew -DsparkVersions=4.1 :iceberg-spark:iceberg-spark-4.1_2.13:jmh \
+ * -PjmhIncludeRegex=IcebergSourceVariantIOBenchmark \
+ * -PjmhOutputPath=build/reports/benchmark/iceberg-source-variant-io-benchmark-result.txt
+ * </code>
+ */
+@Fork(1)
+@Warmup(iterations = 4)
+@Measurement(iterations = 8)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Timeout(time = 10, timeUnit = TimeUnit.MINUTES)
+public class IcebergSourceVariantIOBenchmark extends IcebergSourceBenchmark {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergSourceVariantIOBenchmark.class);
+
+ private static final int NUM_FILES = 4;
+ private static final int NUM_ROWS_PER_FILE = 250_000;
+
+ public static final String COL_ID = "id";
+
+ /** Name of the Spark temp view registered in {@link #setupBenchmark()}. */
+ private static final String TEMP_VIEW = "variant_table";
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, COL_ID, Types.LongType.get()),
+ required(2, "category", Types.IntegerType.get()),
+ required(3, "nested", Types.VariantType.get())); /* The variant. */
+
+ /** Filter to use when filtering on category column: {@value}. */
+ private static final String FILTER_ON_CATEGORY = "category = 5";
+
+ /** Get the category column from the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_CATEGORY =
+ "variant_get(nested, '$.varcategory', 'int')";
+
+ /** Get the ID field from inside the variant: {@value}. */
+ private static final String VARIANT_GET_NESTED_ID = "variant_get(nested, '$.varid', 'int')";
+
+ /** Equivalent of {@link #FILTER_ON_CATEGORY} using the field within the variant: {@value}.. */
+ private static final String FILTER_ON_NESTED_CATEGORY = VARIANT_GET_NESTED_CATEGORY + " = 5";
+
+ /** Table to use in benchmark. */
+ public enum TableType {
+ /** Parquet, no shedding. */
+ Unshredded("parquet", FileFormat.PARQUET, false),
+ /** Parquet, shedded. */
+ Shredded("parquet", FileFormat.PARQUET, true),
+ /** Avro. */
+ Avro("avro", FileFormat.AVRO, false);
+ private final String name;
+ private final FileFormat format;
+ private final boolean shredded;
+
+ TableType(String name, FileFormat format, boolean shredded) {
+ this.name = name;
+ this.format = format;
+ this.shredded = shredded;
+ }
+
+ @Override
+ public String toString() {
+ return "TableType{"
+ + "name='"
+ + name
+ + '\''
+ + ", format="
+ + format
+ + ", shredded="
+ + shredded
+ + '}';
+ }
+ }
+
+ /** Table to use in benchmark. */
+ @Param({"Avro", "Unshredded", "Shredded"})
+ private TableType tableType;
+
+ /** Size must be 20 as the choice of string is derived from category. */
+ private String[] repeatedStrings;
+
+ /** Variant metadata. */
+ private VariantMetadata variantMetadata;
+
+ /** Function to shred variants; will be different for shredded vs unshredded tables. */
+ private VariantShreddingFunction shredder;
+
+ @Override
+ protected Configuration initHadoopConf() {
+ final Configuration conf = new Configuration();
+ conf.setBoolean("fs.file.checksum.verify", false);
+ return conf;
+ }
+
+ /**
+ * This gets invoked in the superclass constructor, at which time the table type is unknown. This
+ * is why the default table type cannot be set.
+ *
+ * @return a table
+ */
+ @Override
+ protected Table initTable() {
+ HadoopTables tables = new HadoopTables(hadoopConf());
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(TableProperties.FORMAT_VERSION, "3");
+ properties.put(TableProperties.SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ // turn off compression to remove it as a factor.
+ properties.put(TableProperties.METADATA_COMPRESSION, "none");
+ properties.put(TableProperties.PARQUET_COMPRESSION, "none");
+ properties.put(TableProperties.AVRO_COMPRESSION, "none");
+ // variant projection pushdown not supported with the vectorized reader.
+ properties.put(TableProperties.PARQUET_VECTORIZATION_ENABLED, "false");
+
+ return tables.create(SCHEMA, PartitionSpec.unpartitioned(), properties, newTableLocation());
+ }
+
+ @Setup
+ public void setupBenchmark() throws IOException {
+ setupSpark();
+
+ // build the parquet schema of the shredded type by creating one variant instance
+ // and type converting it.
+ variantMetadata = Variants.metadata("idstr", "varid", "varcategory", "col4");
+ Type shreddedType =
+ ParquetVariantUtil.toParquetSchema(buildVariant(variantMetadata, 0, 0, "").value());
+ shredder =
+ tableType.shredded ? (fieldId, fieldName) -> shreddedType : (fieldId, fieldName) -> null;
+
+ // algorithm for category produces a value 0..19
+ final int categoryCount = 20;
+ repeatedStrings = new String[categoryCount];
+ IntStream.range(0, categoryCount)
+ .forEach(i -> repeatedStrings[i] = "Longer repeated string " + i);
+ // only one table is populated, to keep setup times down
+ LOG.info("building table {}", tableType);
+ long size =
+ switch (tableType) {
+ case Unshredded, Shredded -> appendVariantData();
+ case Avro -> appendAvroVariantData();
+ };
+ tableDataset().createOrReplaceTempView(TEMP_VIEW);
+ final String summary = String.format(Locale.ROOT, "Size of Table %s : %,3d", tableType, size);
+ LOG.info("{}", summary);
+ tableDataset().printSchema();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Override
+ protected FileFormat fileFormat() {
+ return tableType.format;
+ }
+
+ /**
+ * Create a Spark dataset for the table being benchmarked.
+ *
+ * @return a dataset reading the appropriate table.
+ */
+ private Dataset<Row> tableDataset() {
+ return spark().read().format("iceberg").load(table().location());
+ }
+
+ /** Read the entire table. */
+ @Benchmark
+ public void count() {
+ project("*");
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * ID.
+ *
+ * <p>This measures classic query performance where columnar formats have an advantage over
+ * row-structured ones, especially as the tables grow in size or row width.
+ */
+ @Benchmark
+ public void filterCatProjectID() {
+ select(COL_ID, FILTER_ON_CATEGORY);
+ }
+
+ /**
+ * Perform a sql select operation.
+ *
+ * @param projection columns to project.
+ * @param where optional WHERE clause
+ * @return the result.
+ */
+ private long select(String projection, String where) {
+ final String text =
+ "SELECT " + projection + " FROM " + TEMP_VIEW + (where != null ? " WHERE " + where : "");
+ LOG.info("{}", text);
+ return materializeNonEmpty(text, spark().sql(text));
+ }
+
+ /**
+ * Perform a sql projection operation.
+ *
+ * @param projection columns to project.
+ * @return the result.
+ */
+ private long project(String projection) {
+ return select(projection, null);
+ }
+
+ /**
+ * Read the rows, filtering on the category field, which is outside the variant, then filter on
+ * the variant ID.
+ */
+ @Benchmark
+ public void filterCatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_CATEGORY);
+ }
+
+ /** Project on ID. */
+ @Benchmark
+ public void projectID() {
+ project(COL_ID);
+ }
+
+ /** Extract the varid field only. */
+ @Benchmark
+ public void projectVarID() {
+ project(VARIANT_GET_NESTED_ID);
+ }
+
+ /**
+ * Read filtering on an element within the variant through {@code variant_get()} then project on
+ * ID.
+ *
+ * <p>Filtering on a column within the variant assesses predicate pushdown: is the whole variant
+ * reconstructed before filtering? or does the {@code variant_get()} get used early.
+ */
+ @Benchmark
+ public void filterVarcatProjectID() {
+ select(COL_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Filter to the category match; no projection. */
+ @Benchmark
+ public void filterCat() {
+ select("*", FILTER_ON_CATEGORY);
+ }
+
+ /** Filter to the varcat match; no projection. */
+ @Benchmark
+ public void filterVarCat() {
+ select("*", FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /** Project on the category column. */
+ @Benchmark
+ public void projectCat() {
+ project("category");
+ }
+
+ /** Project on the category field within the variant; no filtering. */
+ @Benchmark
+ public void projectVarCat() {
+ project(VARIANT_GET_NESTED_CATEGORY);
+ }
+
+ /** Filter on nested category field then project on the nested ID field. */
+ @Benchmark
+ public void filterVarcatProjectVarID() {
+ select(VARIANT_GET_NESTED_ID, FILTER_ON_NESTED_CATEGORY);
+ }
+
+ /**
+ * Materialize a dataset by counting its elements; raise an exception if the set is empty, as that
+ * is interpreted a sign the query is somehow broken.
+ *
+ * @param operation operation (for logging)
+ * @param ds dataset
+ * @return count of records returned.
+ */
+ private long materializeNonEmpty(String operation, Dataset<?> ds) {
+ LOG.info("{} table={}", operation, tableType);
+ final long count = ds.count();
Review Comment:
Spark can count records without evaluating projection so it's not really testing the projection here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-4566950412
Marking as ready for review to the extent that the current test dataset shows speedup when ran atop the separate PRs needed for predicate pushdown on shredded variant fields.
here each sql query is run with avro, unshredded and shredded variants:
<img width="346" height="51" alt="Screenshot 2026-05-28 at 19 08 52" src="https://github.com/user-attachments/assets/1a977563-5253-40db-b6e3-6f9c8c43411a" />
before: note how shredded is pathologically worse
<img width="1307" height="1114" alt="Screenshot 2026-05-28 at 19 06 48" src="https://github.com/user-attachments/assets/361c755d-c12e-46db-b247-5322a6ad446a" />
after: note how shredded variants are faster than unshredded for filtering operations.
<img width="1275" height="972" alt="Screenshot 2026-05-28 at 19 08 15" src="https://github.com/user-attachments/assets/a1001c72-7882-4e7e-bb54-76b1328bfc09" />
Special highlight for filtering on an integer field where the value is out of range or not in the IN membership of a query. These only examine the metrics and can skip reading any of the values.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Spark: Add JMH benchmarks for Variants [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #15629:
URL: https://github.com/apache/iceberg/pull/15629#issuecomment-5393543833
I'm not set up to do any java dev right now, so lets target the next release. The big source of merge pain will be the move to spark 4.2; when that goes in the pr will need the spark oart moved, but it will be straightforward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org