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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
