From 8615c094304fb67c7df1d4bb7c1b9258bcaf17d6 Mon Sep 17 00:00:00 2001 From: yufeigu Date: Fri, 12 Nov 2021 22:24:56 -0800 Subject: [PATCH 1/8] Spark: Support vectorized reads with equality deletes --- .../org/apache/iceberg/data/DeleteFilter.java | 15 +++++ .../data/vectorized/ColumnarBatchReader.java | 63 ++++++++++++++++--- .../iceberg/spark/source/BatchDataReader.java | 15 ++++- .../iceberg/spark/source/SparkBatchScan.java | 4 +- 4 files changed, 85 insertions(+), 12 deletions(-) diff --git a/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java b/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java index cf261720a26f..83cb4d38180d 100644 --- a/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java +++ b/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java @@ -71,6 +71,7 @@ public abstract class DeleteFilter { private final Accessor posAccessor; private PositionDeleteIndex deleteRowPositions = null; + private Predicate eqDeleteRows = null; protected DeleteFilter(FileScanTask task, Schema tableSchema, Schema requestedSchema) { this.setFilterThreshold = DEFAULT_SET_FILTER_THRESHOLD; @@ -105,6 +106,10 @@ public boolean hasPosDeletes() { return !posDeletes.isEmpty(); } + public boolean hasEqDeletes() { + return !eqDeletes.isEmpty(); + } + Accessor posAccessor() { return posAccessor; } @@ -192,6 +197,16 @@ protected boolean shouldKeep(T item) { return remainingRowsFilter.filter(records); } + public Predicate eqDeletedRows() { + if (eqDeleteRows == null) { + eqDeleteRows = applyEqDeletes().stream() + .map(Predicate::negate) + .reduce(Predicate::and) + .orElse(t -> true); + } + return eqDeleteRows; + } + public PositionDeleteIndex deletedRowPositions() { if (posDeletes.isEmpty()) { return null; diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java index cc4858f1d61b..51de40ae587a 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java @@ -19,8 +19,10 @@ package org.apache.iceberg.spark.data.vectorized; +import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.function.Predicate; import org.apache.iceberg.arrow.vectorized.BaseBatchReader; import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader; import org.apache.iceberg.data.DeleteFilter; @@ -69,6 +71,13 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { } Pair rowIdMapping = rowIdMapping(numRowsToRead); + int[] eqDeleteRowIdMapping = null; + if(rowIdMapping == null && deletes != null && deletes.hasEqDeletes()) { + eqDeleteRowIdMapping = new int[numRowsToRead]; + for (int i = 0; i < numRowsToRead; i++) { + eqDeleteRowIdMapping[i] = i; + } + } for (int i = 0; i < readers.length; i += 1) { vectorHolders[i] = readers[i].read(vectorHolders[i], numRowsToRead); @@ -78,13 +87,7 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { "Number of rows in the vector %s didn't match expected %s ", numRowsInVector, numRowsToRead); - if (rowIdMapping == null) { - arrowColumnVectors[i] = IcebergArrowColumnVector.forHolder(vectorHolders[i], numRowsInVector); - } else { - int[] rowIdMap = rowIdMapping.first(); - Integer numRows = rowIdMapping.second(); - arrowColumnVectors[i] = ColumnVectorWithFilter.forHolder(vectorHolders[i], rowIdMap, numRows); - } + arrowColumnVectors[i] = arrowColumnVector(rowIdMapping, i, numRowsInVector, eqDeleteRowIdMapping); } rowStartPosInBatch += numRowsToRead; @@ -96,9 +99,26 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { Integer numRows = rowIdMapping.second(); batch.setNumRows(numRows); } + + if (deletes != null && deletes.hasEqDeletes()) { + applyEqDelete(batch, rowIdMapping == null ? null : rowIdMapping.first(), eqDeleteRowIdMapping); + } return batch; } + private ColumnVector arrowColumnVector(Pair rowIdMapping, int index, int numRowsInVector, int[] eqDeleteRowIdMapping) { + if (rowIdMapping != null) { + int[] rowIdMap = rowIdMapping.first(); + Integer numRows = rowIdMapping.second(); + return ColumnVectorWithFilter.forHolder(vectorHolders[index], rowIdMap, numRows); + } else if (deletes != null && deletes.hasEqDeletes()) { + Preconditions.checkArgument(eqDeleteRowIdMapping != null, "Equality delete row Id mapping cannot be null"); + return ColumnVectorWithFilter.forHolder(vectorHolders[index], eqDeleteRowIdMapping, numRowsInVector); + } else { + return IcebergArrowColumnVector.forHolder(vectorHolders[index], numRowsInVector); + } + } + private Pair rowIdMapping(int numRows) { if (deletes != null && deletes.hasPosDeletes()) { return buildRowIdMapping(deletes.deletedRowPositions(), numRows); @@ -137,4 +157,33 @@ private Pair buildRowIdMapping(PositionDeleteIndex deletedRowPos return Pair.of(rowIdMapping, currentRowId); } } + + /** + * Reuse the row Id mapping array to filter out equality deleted rows. + */ + private void applyEqDelete(ColumnarBatch batch, int[] posDeleteRowIdMapping, int[] eqDeleteRowIdMapping) { + int[] rowIdMapping = posDeleteRowIdMapping == null ? eqDeleteRowIdMapping : posDeleteRowIdMapping; + Preconditions.checkArgument(rowIdMapping != null, "Row Id mapping cannot be null"); + int numRows = batch.numRows(); + + Predicate eqDeletedRows = deletes.eqDeletedRows(); + Iterator it = batch.rowIterator(); + int rowId = 0; + int currentRowId = 0; + while (it.hasNext()) { + InternalRow row = it.next(); + if (!eqDeletedRows.test(row)) { + // the row is deleted + numRows--; + } else { + // skip deleted rows by pointing to the next undeleted row Id + rowIdMapping[currentRowId] = rowIdMapping[rowId]; + currentRowId++; + } + + rowId++; + } + + batch.setNumRows(numRows); + } } diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java index 5f05c55789ed..850e91a129e5 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java @@ -76,11 +76,14 @@ CloseableIterator open(FileScanTask task) { Preconditions.checkNotNull(location, "Could not find InputFile associated with FileScanTask"); if (task.file().format() == FileFormat.PARQUET) { SparkDeleteFilter deleteFilter = deleteFilter(task); + // get required schema for filtering out equality-delete rows in case equality-delete uses columns are + // not selected. + Schema requiredSchema = requiredSchema(deleteFilter); Parquet.ReadBuilder builder = Parquet.read(location) - .project(expectedSchema) + .project(requiredSchema) .split(task.start(), task.length()) - .createBatchedReaderFunc(fileSchema -> VectorizedSparkParquetReaders.buildReader(expectedSchema, + .createBatchedReaderFunc(fileSchema -> VectorizedSparkParquetReaders.buildReader(requiredSchema, fileSchema, /* setArrowValidityVector */ NullCheckingForGet.NULL_CHECKING_ENABLED, idToConstant, deleteFilter)) .recordsPerBatch(batchSize) @@ -126,6 +129,14 @@ private SparkDeleteFilter deleteFilter(FileScanTask task) { return task.deletes().isEmpty() ? null : new SparkDeleteFilter(task, table().schema(), expectedSchema); } + private Schema requiredSchema(DeleteFilter deleteFilter) { + if (deleteFilter != null && deleteFilter.hasEqDeletes()) { + return deleteFilter.requiredSchema(); + } else { + return expectedSchema; + } + } + private class SparkDeleteFilter extends DeleteFilter { private final InternalRowWrapper asStructLike; diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java index 81fee430833d..12643be69d2b 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java @@ -172,13 +172,11 @@ public PartitionReaderFactory createReaderFactory() { boolean hasNoDeleteFiles = tasks().stream().noneMatch(TableScanUtil::hasDeletes); - boolean hasNoEqDeleteFiles = tasks().stream().noneMatch(TableScanUtil::hasEqDeletes); - boolean batchReadsEnabled = batchReadsEnabled(allParquetFileScanTasks, allOrcFileScanTasks); boolean batchReadOrc = hasNoDeleteFiles && allOrcFileScanTasks; - boolean batchReadParquet = hasNoEqDeleteFiles && allParquetFileScanTasks && atLeastOneColumn && onlyPrimitives; + boolean batchReadParquet = allParquetFileScanTasks && atLeastOneColumn && onlyPrimitives; boolean readUsingBatch = batchReadsEnabled && (batchReadOrc || batchReadParquet); From b680dc1c5f0aea4d5685ce20565d021e7d414ad3 Mon Sep 17 00:00:00 2001 From: yufeigu Date: Mon, 15 Nov 2021 12:33:13 -0800 Subject: [PATCH 2/8] Simplify the method applyEqDelete --- .../spark/data/vectorized/ColumnarBatchReader.java | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java index 51de40ae587a..9df2860e40ff 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java @@ -164,7 +164,6 @@ private Pair buildRowIdMapping(PositionDeleteIndex deletedRowPos private void applyEqDelete(ColumnarBatch batch, int[] posDeleteRowIdMapping, int[] eqDeleteRowIdMapping) { int[] rowIdMapping = posDeleteRowIdMapping == null ? eqDeleteRowIdMapping : posDeleteRowIdMapping; Preconditions.checkArgument(rowIdMapping != null, "Row Id mapping cannot be null"); - int numRows = batch.numRows(); Predicate eqDeletedRows = deletes.eqDeletedRows(); Iterator it = batch.rowIterator(); @@ -172,10 +171,8 @@ private void applyEqDelete(ColumnarBatch batch, int[] posDeleteRowIdMapping, int int currentRowId = 0; while (it.hasNext()) { InternalRow row = it.next(); - if (!eqDeletedRows.test(row)) { - // the row is deleted - numRows--; - } else { + if (eqDeletedRows.test(row)) { + // the row is NOT deleted // skip deleted rows by pointing to the next undeleted row Id rowIdMapping[currentRowId] = rowIdMapping[rowId]; currentRowId++; @@ -184,6 +181,6 @@ private void applyEqDelete(ColumnarBatch batch, int[] posDeleteRowIdMapping, int rowId++; } - batch.setNumRows(numRows); + batch.setNumRows(currentRowId); } } From bded78acdcce70cc19546ccc8b416cd6e6d22b41 Mon Sep 17 00:00:00 2001 From: yufeigu Date: Mon, 15 Nov 2021 12:40:18 -0800 Subject: [PATCH 3/8] Fix style issue --- .../data/vectorized/ColumnarBatchReader.java | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java index 9df2860e40ff..d2171e560bdb 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java @@ -36,6 +36,7 @@ import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.vectorized.ColumnVector; import org.apache.spark.sql.vectorized.ColumnarBatch; +import org.jetbrains.annotations.Nullable; /** * {@link VectorizedReader} that returns Spark's {@link ColumnarBatch} to support Spark's vectorized read path. The @@ -71,13 +72,7 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { } Pair rowIdMapping = rowIdMapping(numRowsToRead); - int[] eqDeleteRowIdMapping = null; - if(rowIdMapping == null && deletes != null && deletes.hasEqDeletes()) { - eqDeleteRowIdMapping = new int[numRowsToRead]; - for (int i = 0; i < numRowsToRead; i++) { - eqDeleteRowIdMapping[i] = i; - } - } + int[] eqDeleteRowIdMapping = initEqDeleteRowIdMapping(numRowsToRead, rowIdMapping); for (int i = 0; i < readers.length; i += 1) { vectorHolders[i] = readers[i].read(vectorHolders[i], numRowsToRead); @@ -106,7 +101,20 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { return batch; } - private ColumnVector arrowColumnVector(Pair rowIdMapping, int index, int numRowsInVector, int[] eqDeleteRowIdMapping) { + @Nullable + private int[] initEqDeleteRowIdMapping(int numRowsToRead, Pair rowIdMapping) { + int[] eqDeleteRowIdMapping = null; + if (rowIdMapping == null && deletes != null && deletes.hasEqDeletes()) { + eqDeleteRowIdMapping = new int[numRowsToRead]; + for (int i = 0; i < numRowsToRead; i++) { + eqDeleteRowIdMapping[i] = i; + } + } + return eqDeleteRowIdMapping; + } + + private ColumnVector arrowColumnVector(Pair rowIdMapping, int index, int numRowsInVector, + int[] eqDeleteRowIdMapping) { if (rowIdMapping != null) { int[] rowIdMap = rowIdMapping.first(); Integer numRows = rowIdMapping.second(); From 0b59ce002c22b0537930de4fefe47231d6128c89 Mon Sep 17 00:00:00 2001 From: yufeigu Date: Fri, 19 Nov 2021 13:19:42 -0800 Subject: [PATCH 4/8] Add benchmark for eq deletes --- .../source/IcebergSourceDeleteBenchmark.java | 53 ++++++++++++++++ ...IcebergSourceParquetEqDeleteBenchmark.java | 60 +++++++++++++++++++ ...ebergSourceParquetPosDeleteBenchmark.java} | 2 +- 3 files changed, 114 insertions(+), 1 deletion(-) create mode 100644 spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java rename spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/{IcebergSourceParquetDeleteBenchmark.java => IcebergSourceParquetPosDeleteBenchmark.java} (95%) diff --git a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceDeleteBenchmark.java b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceDeleteBenchmark.java index d5a6db3b385c..82c64b707c07 100644 --- a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceDeleteBenchmark.java +++ b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceDeleteBenchmark.java @@ -33,15 +33,18 @@ import org.apache.iceberg.TableProperties; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.io.ClusteredEqualityDeleteWriter; import org.apache.iceberg.io.ClusteredPositionDeleteWriter; import org.apache.iceberg.io.OutputFileFactory; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Types; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.TearDown; @@ -190,6 +193,56 @@ writerFactory, fileFactory, table().io(), rowDelta.validateDeletedFiles().commit(); } + protected void writeEqDeletes(long numRows, double percentage) throws IOException { + Set deletedValues = Sets.newHashSet(); + while (deletedValues.size() < numRows * percentage) { + deletedValues.add(ThreadLocalRandom.current().nextLong(numRows)); + } + + List rows = Lists.newArrayList(); + for (Long value : deletedValues) { + GenericInternalRow genericInternalRow = new GenericInternalRow(7); + genericInternalRow.setLong(0, value); + genericInternalRow.setInt(1, (int) (value % Integer.MAX_VALUE)); + genericInternalRow.setFloat(2, (float) value); + genericInternalRow.setNullAt(3); + genericInternalRow.setNullAt(4); + genericInternalRow.setNullAt(5); + genericInternalRow.setNullAt(6); + rows.add(genericInternalRow); + } + LOG.info("Num of equality deleted rows: {}", rows.size()); + + writeEqDeletes(rows); + } + + private void writeEqDeletes(List rows) throws IOException { + int equalityFieldId = table().schema().findField("longCol").fieldId(); + + OutputFileFactory fileFactory = newFileFactory(); + SparkFileWriterFactory writerFactory = SparkFileWriterFactory + .builderFor(table()) + .dataFileFormat(fileFormat()) + .equalityDeleteRowSchema(table().schema()) + .equalityFieldIds(new int[]{equalityFieldId}) + .build(); + + ClusteredEqualityDeleteWriter writer = new ClusteredEqualityDeleteWriter<>( + writerFactory, fileFactory, table().io(), fileFormat(), TARGET_FILE_SIZE_IN_BYTES); + + PartitionSpec unpartitionedSpec = table().specs().get(0); + try (ClusteredEqualityDeleteWriter closeableWriter = writer) { + for (InternalRow row : rows) { + closeableWriter.write(row, unpartitionedSpec, null); + } + } + + RowDelta rowDelta = table().newRowDelta(); + LOG.info("Num of Delete File: {}", writer.result().deleteFiles().size()); + writer.result().deleteFiles().forEach(rowDelta::addDeletes); + rowDelta.validateDeletedFiles().commit(); + } + private OutputFileFactory newFileFactory() { return OutputFileFactory.builderFor(table(), 1, 1) .format(fileFormat()) diff --git a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java new file mode 100644 index 000000000000..adf6fea85a0e --- /dev/null +++ b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java @@ -0,0 +1,60 @@ +/* + * 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.spark.source.parquet; + +import java.io.IOException; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.spark.source.IcebergSourceDeleteBenchmark; +import org.openjdk.jmh.annotations.Param; + +/** + * A benchmark that evaluates the non-vectorized read and vectorized read with pos-delete in the Spark data source for + * Iceberg. + *

+ * This class uses a dataset with a flat schema. + * To run this benchmark for spark-3.2: + * + * ./gradlew :iceberg-spark:iceberg-spark-3.2:jmh + * -PjmhIncludeRegex=IcebergSourceParquetEqDeleteBenchmark + * -PjmhOutputPath=benchmark/iceberg-source-parquet-eq-delete-benchmark-result.txt + * + */ +public class IcebergSourceParquetEqDeleteBenchmark extends IcebergSourceDeleteBenchmark { + @Param({"0", "0.000001", "0.05", "0.25", "0.5", "1"}) + private double percentDeleteRow; + + @Override + protected void appendData() throws IOException { + for (int fileNum = 1; fileNum <= NUM_FILES; fileNum++) { + writeData(fileNum); + + if (percentDeleteRow > 0) { + // add equality deletes + table().refresh(); + writeEqDeletes(NUM_ROWS, percentDeleteRow); + } + } + } + + @Override + protected FileFormat fileFormat() { + return FileFormat.PARQUET; + } +} diff --git a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetDeleteBenchmark.java b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetPosDeleteBenchmark.java similarity index 95% rename from spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetDeleteBenchmark.java rename to spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetPosDeleteBenchmark.java index 234c6c5666ca..ab231b10b2d9 100644 --- a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetDeleteBenchmark.java +++ b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetPosDeleteBenchmark.java @@ -37,7 +37,7 @@ * -PjmhOutputPath=benchmark/iceberg-source-parquet-delete-benchmark-result.txt * */ -public class IcebergSourceParquetDeleteBenchmark extends IcebergSourceDeleteBenchmark { +public class IcebergSourceParquetPosDeleteBenchmark extends IcebergSourceDeleteBenchmark { @Param({"0", "0.000001", "0.05", "0.25", "0.5", "1"}) private double percentDeleteRow; From 538e919d51207a008b09dd7bc264421e24884b5f Mon Sep 17 00:00:00 2001 From: yufeigu Date: Fri, 19 Nov 2021 13:21:33 -0800 Subject: [PATCH 5/8] Add benchmark for eq deletes --- .../source/parquet/IcebergSourceParquetEqDeleteBenchmark.java | 4 ++-- .../parquet/IcebergSourceParquetPosDeleteBenchmark.java | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java index adf6fea85a0e..a2cfd56bd2b4 100644 --- a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java +++ b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetEqDeleteBenchmark.java @@ -25,8 +25,8 @@ import org.openjdk.jmh.annotations.Param; /** - * A benchmark that evaluates the non-vectorized read and vectorized read with pos-delete in the Spark data source for - * Iceberg. + * A benchmark that evaluates the non-vectorized read and vectorized read with equality delete in the Spark data source + * for Iceberg. *

* This class uses a dataset with a flat schema. * To run this benchmark for spark-3.2: diff --git a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetPosDeleteBenchmark.java b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetPosDeleteBenchmark.java index ab231b10b2d9..acfc86ecd6f1 100644 --- a/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetPosDeleteBenchmark.java +++ b/spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceParquetPosDeleteBenchmark.java @@ -33,8 +33,8 @@ * To run this benchmark for spark-3.2: * * ./gradlew :iceberg-spark:iceberg-spark-3.2:jmh - * -PjmhIncludeRegex=IcebergSourceParquetDeleteBenchmark - * -PjmhOutputPath=benchmark/iceberg-source-parquet-delete-benchmark-result.txt + * -PjmhIncludeRegex=IcebergSourceParquetPosDeleteBenchmark + * -PjmhOutputPath=benchmark/iceberg-source-parquet-pos-delete-benchmark-result.txt * */ public class IcebergSourceParquetPosDeleteBenchmark extends IcebergSourceDeleteBenchmark { From 800c4b2907599a47d74efe9472db47e1679efef5 Mon Sep 17 00:00:00 2001 From: yufeigu Date: Fri, 19 Nov 2021 13:49:23 -0800 Subject: [PATCH 6/8] Resolve comments. --- .../org/apache/iceberg/data/DeleteFilter.java | 2 +- .../data/vectorized/ColumnarBatchReader.java | 30 ++++++++++--------- 2 files changed, 17 insertions(+), 15 deletions(-) diff --git a/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java b/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java index 83cb4d38180d..6901859600f4 100644 --- a/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java +++ b/data/src/main/java/org/apache/iceberg/data/DeleteFilter.java @@ -197,7 +197,7 @@ protected boolean shouldKeep(T item) { return remainingRowsFilter.filter(records); } - public Predicate eqDeletedRows() { + public Predicate eqDeletedRowFilter() { if (eqDeleteRows == null) { eqDeleteRows = applyEqDeletes().stream() .map(Predicate::negate) diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java index d2171e560bdb..841a7091bded 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java @@ -22,7 +22,6 @@ import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.function.Predicate; import org.apache.iceberg.arrow.vectorized.BaseBatchReader; import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader; import org.apache.iceberg.data.DeleteFilter; @@ -71,8 +70,8 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { closeVectors(); } - Pair rowIdMapping = rowIdMapping(numRowsToRead); - int[] eqDeleteRowIdMapping = initEqDeleteRowIdMapping(numRowsToRead, rowIdMapping); + Pair posDeleteRowIdMapping = rowIdMapping(numRowsToRead); + int[] eqDeleteRowIdMapping = initEqDeleteRowIdMapping(numRowsToRead, posDeleteRowIdMapping); for (int i = 0; i < readers.length; i += 1) { vectorHolders[i] = readers[i].read(vectorHolders[i], numRowsToRead); @@ -82,22 +81,29 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { "Number of rows in the vector %s didn't match expected %s ", numRowsInVector, numRowsToRead); - arrowColumnVectors[i] = arrowColumnVector(rowIdMapping, i, numRowsInVector, eqDeleteRowIdMapping); + arrowColumnVectors[i] = arrowColumnVector(posDeleteRowIdMapping, i, numRowsInVector, eqDeleteRowIdMapping); } rowStartPosInBatch += numRowsToRead; ColumnarBatch batch = new ColumnarBatch(arrowColumnVectors); - if (rowIdMapping == null) { + if (posDeleteRowIdMapping == null) { batch.setNumRows(numRowsToRead); } else { - Integer numRows = rowIdMapping.second(); + Integer numRows = posDeleteRowIdMapping.second(); batch.setNumRows(numRows); } if (deletes != null && deletes.hasEqDeletes()) { - applyEqDelete(batch, rowIdMapping == null ? null : rowIdMapping.first(), eqDeleteRowIdMapping); + int[] rowIdMapping = eqDeleteRowIdMapping; + if (posDeleteRowIdMapping != null && posDeleteRowIdMapping.first() != null) { + rowIdMapping = posDeleteRowIdMapping.first(); + } + Preconditions.checkArgument(rowIdMapping != null, "Row Id mapping cannot be null"); + + applyEqDelete(batch, rowIdMapping); } + return batch; } @@ -167,19 +173,15 @@ private Pair buildRowIdMapping(PositionDeleteIndex deletedRowPos } /** - * Reuse the row Id mapping array to filter out equality deleted rows. + * Reuse the row id mapping array to filter out equality deleted rows. */ - private void applyEqDelete(ColumnarBatch batch, int[] posDeleteRowIdMapping, int[] eqDeleteRowIdMapping) { - int[] rowIdMapping = posDeleteRowIdMapping == null ? eqDeleteRowIdMapping : posDeleteRowIdMapping; - Preconditions.checkArgument(rowIdMapping != null, "Row Id mapping cannot be null"); - - Predicate eqDeletedRows = deletes.eqDeletedRows(); + private void applyEqDelete(ColumnarBatch batch, int[] rowIdMapping) { Iterator it = batch.rowIterator(); int rowId = 0; int currentRowId = 0; while (it.hasNext()) { InternalRow row = it.next(); - if (eqDeletedRows.test(row)) { + if (deletes.eqDeletedRowFilter().test(row)) { // the row is NOT deleted // skip deleted rows by pointing to the next undeleted row Id rowIdMapping[currentRowId] = rowIdMapping[rowId]; From b1e9c8cb95c3812a652fbe5bcae5e107bd6285a9 Mon Sep 17 00:00:00 2001 From: yufeigu Date: Wed, 8 Dec 2021 18:15:27 -0800 Subject: [PATCH 7/8] Refactor --- .../vectorized/ColumnBatchWithRowMapping.java | 156 ++++++++++++++++++ .../data/vectorized/ColumnarBatchReader.java | 118 +------------ 2 files changed, 161 insertions(+), 113 deletions(-) create mode 100644 spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java new file mode 100644 index 000000000000..6f440b0fa032 --- /dev/null +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java @@ -0,0 +1,156 @@ +/* + * 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.spark.data.vectorized; + +import java.util.Iterator; +import org.apache.iceberg.data.DeleteFilter; +import org.apache.iceberg.deletes.PositionDeleteIndex; +import org.apache.iceberg.util.Pair; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.vectorized.ColumnVector; +import org.apache.spark.sql.vectorized.ColumnarBatch; + +class ColumnBatchWithRowMapping { + private final DeleteFilter deletes; + private int[] rowIdMapping; // the rowId mapping to skip deleted rows for all column vectors inside a batch + private int numRows; + private ColumnarBatch batch; + + ColumnBatchWithRowMapping(DeleteFilter deletes, int numRowsToRead, long rowStartPosInBatch) { + this.deletes = deletes; + initRowIdMapping(numRowsToRead, rowStartPosInBatch); + } + + ColumnarBatch createColumnBatch(ColumnVector[] columns) { + batch = new ColumnarBatch(columns); + batch.setNumRows(numRows); + + if (hasEqDeletes()) { + applyEqDelete(); + } + return batch; + } + + boolean hasDeletes() { + return rowIdMapping != null; + } + + private boolean hasEqDeletes() { + return deletes != null && deletes.hasEqDeletes(); + } + + int[] rowIdMapping() { + return rowIdMapping; + } + + int numRows() { + return numRows; + } + + private void initRowIdMapping(int numRowsToRead, long rowStartPosInBatch) { + Pair posDeleteRowIdMapping = posDelRowIdMapping(numRowsToRead, rowStartPosInBatch); + if (posDeleteRowIdMapping != null) { + rowIdMapping = posDeleteRowIdMapping.first(); + numRows = posDeleteRowIdMapping.second(); + } else { + numRows = numRowsToRead; + rowIdMapping = initEqDeleteRowIdMapping(numRowsToRead); + } + } + + private Pair posDelRowIdMapping(int numRowsToRead, long rowStartPosInBatch) { + if (deletes != null && deletes.hasPosDeletes()) { + return buildPosDelRowIdMapping(deletes.deletedRowPositions(), numRowsToRead, rowStartPosInBatch); + } else { + return null; + } + } + + /** + * Build a row id mapping inside a batch, which skips delete rows. For example, if the 1st and 3rd rows are deleted in + * a batch with 5 rows, the mapping would be {0->1, 1->3, 2->4}, and the new num of rows is 3. + * + * @param deletedRowPositions a set of deleted row positions + * @param numRowsToRead the num of rows + * @return the mapping array and the new num of rows in a batch, null if no row is deleted + */ + private Pair buildPosDelRowIdMapping(PositionDeleteIndex deletedRowPositions, int numRowsToRead, + long rowStartPosInBatch) { + if (deletedRowPositions == null) { + return null; + } + + int[] posDelRowIdMapping = new int[numRowsToRead]; + int originalRowId = 0; + int currentRowId = 0; + while (originalRowId < numRowsToRead) { + if (!deletedRowPositions.deleted(originalRowId + rowStartPosInBatch)) { + posDelRowIdMapping[currentRowId] = originalRowId; + currentRowId++; + } + originalRowId++; + } + + if (currentRowId == numRowsToRead) { + // there is no delete in this batch + return null; + } else { + return Pair.of(posDelRowIdMapping, currentRowId); + } + } + + private int[] initEqDeleteRowIdMapping(int numRowsToRead) { + int[] eqDeleteRowIdMapping = null; + if (hasEqDeletes()) { + eqDeleteRowIdMapping = new int[numRowsToRead]; + for (int i = 0; i < numRowsToRead; i++) { + eqDeleteRowIdMapping[i] = i; + } + } + return eqDeleteRowIdMapping; + } + + /** + * Filter out the equality deleted rows. Here is an example, + * [0,1,2,3,4,5,6,7] -- Original + * POS Delete 2, 6 + * [0,1,3,4,5,7,-,-] -- After Apply Pos Deletes [Set Num records to 6] + * Equality delete 1 <= x <= 3 + * [0,4,5,7,-,-,-,-] -- After Apply Eq Deletes [Set Num records to 4] + */ + private void applyEqDelete() { + Iterator it = batch.rowIterator(); + int rowId = 0; + int currentRowId = 0; + while (it.hasNext()) { + InternalRow row = it.next(); + if (deletes.eqDeletedRowFilter().test(row)) { + // the row is NOT deleted + // skip deleted rows by pointing to the next undeleted row Id + rowIdMapping[currentRowId] = rowIdMapping[rowId]; + currentRowId++; + } + + rowId++; + } + + batch.setNumRows(currentRowId); + } +} diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java index 841a7091bded..be4c1f413ec1 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java @@ -19,23 +19,19 @@ package org.apache.iceberg.spark.data.vectorized; -import java.util.Iterator; import java.util.List; import java.util.Map; import org.apache.iceberg.arrow.vectorized.BaseBatchReader; import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader; import org.apache.iceberg.data.DeleteFilter; -import org.apache.iceberg.deletes.PositionDeleteIndex; import org.apache.iceberg.parquet.VectorizedReader; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; -import org.apache.iceberg.util.Pair; import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.ColumnPath; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.vectorized.ColumnVector; import org.apache.spark.sql.vectorized.ColumnarBatch; -import org.jetbrains.annotations.Nullable; /** * {@link VectorizedReader} that returns Spark's {@link ColumnarBatch} to support Spark's vectorized read path. The @@ -70,8 +66,7 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { closeVectors(); } - Pair posDeleteRowIdMapping = rowIdMapping(numRowsToRead); - int[] eqDeleteRowIdMapping = initEqDeleteRowIdMapping(numRowsToRead, posDeleteRowIdMapping); + ColumnBatchWithRowMapping batch = new ColumnBatchWithRowMapping(deletes, numRowsToRead, rowStartPosInBatch); for (int i = 0; i < readers.length; i += 1) { vectorHolders[i] = readers[i].read(vectorHolders[i], numRowsToRead); @@ -81,116 +76,13 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { "Number of rows in the vector %s didn't match expected %s ", numRowsInVector, numRowsToRead); - arrowColumnVectors[i] = arrowColumnVector(posDeleteRowIdMapping, i, numRowsInVector, eqDeleteRowIdMapping); + arrowColumnVectors[i] = batch.hasDeletes() ? + ColumnVectorWithFilter.forHolder(vectorHolders[i], batch.rowIdMapping(), batch.numRows()) : + IcebergArrowColumnVector.forHolder(vectorHolders[i], numRowsInVector); } rowStartPosInBatch += numRowsToRead; - ColumnarBatch batch = new ColumnarBatch(arrowColumnVectors); - if (posDeleteRowIdMapping == null) { - batch.setNumRows(numRowsToRead); - } else { - Integer numRows = posDeleteRowIdMapping.second(); - batch.setNumRows(numRows); - } - - if (deletes != null && deletes.hasEqDeletes()) { - int[] rowIdMapping = eqDeleteRowIdMapping; - if (posDeleteRowIdMapping != null && posDeleteRowIdMapping.first() != null) { - rowIdMapping = posDeleteRowIdMapping.first(); - } - Preconditions.checkArgument(rowIdMapping != null, "Row Id mapping cannot be null"); - - applyEqDelete(batch, rowIdMapping); - } - - return batch; - } - - @Nullable - private int[] initEqDeleteRowIdMapping(int numRowsToRead, Pair rowIdMapping) { - int[] eqDeleteRowIdMapping = null; - if (rowIdMapping == null && deletes != null && deletes.hasEqDeletes()) { - eqDeleteRowIdMapping = new int[numRowsToRead]; - for (int i = 0; i < numRowsToRead; i++) { - eqDeleteRowIdMapping[i] = i; - } - } - return eqDeleteRowIdMapping; - } - - private ColumnVector arrowColumnVector(Pair rowIdMapping, int index, int numRowsInVector, - int[] eqDeleteRowIdMapping) { - if (rowIdMapping != null) { - int[] rowIdMap = rowIdMapping.first(); - Integer numRows = rowIdMapping.second(); - return ColumnVectorWithFilter.forHolder(vectorHolders[index], rowIdMap, numRows); - } else if (deletes != null && deletes.hasEqDeletes()) { - Preconditions.checkArgument(eqDeleteRowIdMapping != null, "Equality delete row Id mapping cannot be null"); - return ColumnVectorWithFilter.forHolder(vectorHolders[index], eqDeleteRowIdMapping, numRowsInVector); - } else { - return IcebergArrowColumnVector.forHolder(vectorHolders[index], numRowsInVector); - } - } - - private Pair rowIdMapping(int numRows) { - if (deletes != null && deletes.hasPosDeletes()) { - return buildRowIdMapping(deletes.deletedRowPositions(), numRows); - } else { - return null; - } - } - - /** - * Build a row id mapping inside a batch, which skips delete rows. For example, if the 1st and 3rd rows are deleted in - * a batch with 5 rows, the mapping would be {0->1, 1->3, 2->4}, and the new num of rows is 3. - * @param deletedRowPositions a set of deleted row positions - * @param numRows the num of rows - * @return the mapping array and the new num of rows in a batch, null if no row is deleted - */ - private Pair buildRowIdMapping(PositionDeleteIndex deletedRowPositions, int numRows) { - if (deletedRowPositions == null) { - return null; - } - - int[] rowIdMapping = new int[numRows]; - int originalRowId = 0; - int currentRowId = 0; - while (originalRowId < numRows) { - if (!deletedRowPositions.deleted(originalRowId + rowStartPosInBatch)) { - rowIdMapping[currentRowId] = originalRowId; - currentRowId++; - } - originalRowId++; - } - - if (currentRowId == numRows) { - // there is no delete in this batch - return null; - } else { - return Pair.of(rowIdMapping, currentRowId); - } - } - - /** - * Reuse the row id mapping array to filter out equality deleted rows. - */ - private void applyEqDelete(ColumnarBatch batch, int[] rowIdMapping) { - Iterator it = batch.rowIterator(); - int rowId = 0; - int currentRowId = 0; - while (it.hasNext()) { - InternalRow row = it.next(); - if (deletes.eqDeletedRowFilter().test(row)) { - // the row is NOT deleted - // skip deleted rows by pointing to the next undeleted row Id - rowIdMapping[currentRowId] = rowIdMapping[rowId]; - currentRowId++; - } - - rowId++; - } - - batch.setNumRows(currentRowId); + return batch.createColumnBatch(arrowColumnVectors); } } From 5eca20d48f327bf2ebe269674cc72f3a3895d397 Mon Sep 17 00:00:00 2001 From: yufeigu Date: Thu, 9 Dec 2021 14:48:51 -0800 Subject: [PATCH 8/8] Resolve comments. --- .../vectorized/ColumnBatchWithRowMapping.java | 156 ----------------- .../data/vectorized/ColumnarBatchReader.java | 159 ++++++++++++++++-- 2 files changed, 143 insertions(+), 172 deletions(-) delete mode 100644 spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java deleted file mode 100644 index 6f440b0fa032..000000000000 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnBatchWithRowMapping.java +++ /dev/null @@ -1,156 +0,0 @@ -/* - * 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.spark.data.vectorized; - -import java.util.Iterator; -import org.apache.iceberg.data.DeleteFilter; -import org.apache.iceberg.deletes.PositionDeleteIndex; -import org.apache.iceberg.util.Pair; -import org.apache.spark.sql.catalyst.InternalRow; -import org.apache.spark.sql.vectorized.ColumnVector; -import org.apache.spark.sql.vectorized.ColumnarBatch; - -class ColumnBatchWithRowMapping { - private final DeleteFilter deletes; - private int[] rowIdMapping; // the rowId mapping to skip deleted rows for all column vectors inside a batch - private int numRows; - private ColumnarBatch batch; - - ColumnBatchWithRowMapping(DeleteFilter deletes, int numRowsToRead, long rowStartPosInBatch) { - this.deletes = deletes; - initRowIdMapping(numRowsToRead, rowStartPosInBatch); - } - - ColumnarBatch createColumnBatch(ColumnVector[] columns) { - batch = new ColumnarBatch(columns); - batch.setNumRows(numRows); - - if (hasEqDeletes()) { - applyEqDelete(); - } - return batch; - } - - boolean hasDeletes() { - return rowIdMapping != null; - } - - private boolean hasEqDeletes() { - return deletes != null && deletes.hasEqDeletes(); - } - - int[] rowIdMapping() { - return rowIdMapping; - } - - int numRows() { - return numRows; - } - - private void initRowIdMapping(int numRowsToRead, long rowStartPosInBatch) { - Pair posDeleteRowIdMapping = posDelRowIdMapping(numRowsToRead, rowStartPosInBatch); - if (posDeleteRowIdMapping != null) { - rowIdMapping = posDeleteRowIdMapping.first(); - numRows = posDeleteRowIdMapping.second(); - } else { - numRows = numRowsToRead; - rowIdMapping = initEqDeleteRowIdMapping(numRowsToRead); - } - } - - private Pair posDelRowIdMapping(int numRowsToRead, long rowStartPosInBatch) { - if (deletes != null && deletes.hasPosDeletes()) { - return buildPosDelRowIdMapping(deletes.deletedRowPositions(), numRowsToRead, rowStartPosInBatch); - } else { - return null; - } - } - - /** - * Build a row id mapping inside a batch, which skips delete rows. For example, if the 1st and 3rd rows are deleted in - * a batch with 5 rows, the mapping would be {0->1, 1->3, 2->4}, and the new num of rows is 3. - * - * @param deletedRowPositions a set of deleted row positions - * @param numRowsToRead the num of rows - * @return the mapping array and the new num of rows in a batch, null if no row is deleted - */ - private Pair buildPosDelRowIdMapping(PositionDeleteIndex deletedRowPositions, int numRowsToRead, - long rowStartPosInBatch) { - if (deletedRowPositions == null) { - return null; - } - - int[] posDelRowIdMapping = new int[numRowsToRead]; - int originalRowId = 0; - int currentRowId = 0; - while (originalRowId < numRowsToRead) { - if (!deletedRowPositions.deleted(originalRowId + rowStartPosInBatch)) { - posDelRowIdMapping[currentRowId] = originalRowId; - currentRowId++; - } - originalRowId++; - } - - if (currentRowId == numRowsToRead) { - // there is no delete in this batch - return null; - } else { - return Pair.of(posDelRowIdMapping, currentRowId); - } - } - - private int[] initEqDeleteRowIdMapping(int numRowsToRead) { - int[] eqDeleteRowIdMapping = null; - if (hasEqDeletes()) { - eqDeleteRowIdMapping = new int[numRowsToRead]; - for (int i = 0; i < numRowsToRead; i++) { - eqDeleteRowIdMapping[i] = i; - } - } - return eqDeleteRowIdMapping; - } - - /** - * Filter out the equality deleted rows. Here is an example, - * [0,1,2,3,4,5,6,7] -- Original - * POS Delete 2, 6 - * [0,1,3,4,5,7,-,-] -- After Apply Pos Deletes [Set Num records to 6] - * Equality delete 1 <= x <= 3 - * [0,4,5,7,-,-,-,-] -- After Apply Eq Deletes [Set Num records to 4] - */ - private void applyEqDelete() { - Iterator it = batch.rowIterator(); - int rowId = 0; - int currentRowId = 0; - while (it.hasNext()) { - InternalRow row = it.next(); - if (deletes.eqDeletedRowFilter().test(row)) { - // the row is NOT deleted - // skip deleted rows by pointing to the next undeleted row Id - rowIdMapping[currentRowId] = rowIdMapping[rowId]; - currentRowId++; - } - - rowId++; - } - - batch.setNumRows(currentRowId); - } -} diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java index be4c1f413ec1..e74a947e835c 100644 --- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java +++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java @@ -19,13 +19,16 @@ package org.apache.iceberg.spark.data.vectorized; +import java.util.Iterator; import java.util.List; import java.util.Map; import org.apache.iceberg.arrow.vectorized.BaseBatchReader; import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader; import org.apache.iceberg.data.DeleteFilter; +import org.apache.iceberg.deletes.PositionDeleteIndex; import org.apache.iceberg.parquet.VectorizedReader; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.util.Pair; import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.ColumnPath; @@ -59,30 +62,154 @@ public void setDeleteFilter(DeleteFilter deleteFilter) { @Override public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) { - Preconditions.checkArgument(numRowsToRead > 0, "Invalid number of rows to read: %s", numRowsToRead); - ColumnVector[] arrowColumnVectors = new ColumnVector[readers.length]; - if (reuse == null) { closeVectors(); } - ColumnBatchWithRowMapping batch = new ColumnBatchWithRowMapping(deletes, numRowsToRead, rowStartPosInBatch); + ColumnBatchLoader batchLoader = new ColumnBatchLoader(numRowsToRead); + rowStartPosInBatch += numRowsToRead; + return batchLoader.columnarBatch; + } - for (int i = 0; i < readers.length; i += 1) { - vectorHolders[i] = readers[i].read(vectorHolders[i], numRowsToRead); - int numRowsInVector = vectorHolders[i].numValues(); - Preconditions.checkState( - numRowsInVector == numRowsToRead, - "Number of rows in the vector %s didn't match expected %s ", numRowsInVector, - numRowsToRead); + private class ColumnBatchLoader { + private int[] rowIdMapping; // the rowId mapping to skip deleted rows for all column vectors inside a batch + private int numRows; + private ColumnarBatch columnarBatch; - arrowColumnVectors[i] = batch.hasDeletes() ? - ColumnVectorWithFilter.forHolder(vectorHolders[i], batch.rowIdMapping(), batch.numRows()) : - IcebergArrowColumnVector.forHolder(vectorHolders[i], numRowsInVector); + ColumnBatchLoader(int numRowsToRead) { + initRowIdMapping(numRowsToRead); + loadDataToColumnBatch(numRowsToRead); } - rowStartPosInBatch += numRowsToRead; + ColumnarBatch loadDataToColumnBatch(int numRowsToRead) { + Preconditions.checkArgument(numRowsToRead > 0, "Invalid number of rows to read: %s", numRowsToRead); + ColumnVector[] arrowColumnVectors = readDataToColumnVectors(numRowsToRead); + + columnarBatch = new ColumnarBatch(arrowColumnVectors); + columnarBatch.setNumRows(numRows); + + if (hasEqDeletes()) { + applyEqDelete(); + } + return columnarBatch; + } + + ColumnVector[] readDataToColumnVectors(int numRowsToRead) { + ColumnVector[] arrowColumnVectors = new ColumnVector[readers.length]; + + for (int i = 0; i < readers.length; i += 1) { + vectorHolders[i] = readers[i].read(vectorHolders[i], numRowsToRead); + int numRowsInVector = vectorHolders[i].numValues(); + Preconditions.checkState( + numRowsInVector == numRowsToRead, + "Number of rows in the vector %s didn't match expected %s ", numRowsInVector, + numRowsToRead); + + arrowColumnVectors[i] = hasDeletes() ? + ColumnVectorWithFilter.forHolder(vectorHolders[i], rowIdMapping, numRows) : + IcebergArrowColumnVector.forHolder(vectorHolders[i], numRowsInVector); + } + return arrowColumnVectors; + } + + boolean hasDeletes() { + return rowIdMapping != null; + } + + boolean hasEqDeletes() { + return deletes != null && deletes.hasEqDeletes(); + } + + void initRowIdMapping(int numRowsToRead) { + Pair posDeleteRowIdMapping = posDelRowIdMapping(numRowsToRead); + if (posDeleteRowIdMapping != null) { + rowIdMapping = posDeleteRowIdMapping.first(); + numRows = posDeleteRowIdMapping.second(); + } else { + numRows = numRowsToRead; + rowIdMapping = initEqDeleteRowIdMapping(numRowsToRead); + } + } + + Pair posDelRowIdMapping(int numRowsToRead) { + if (deletes != null && deletes.hasPosDeletes()) { + return buildPosDelRowIdMapping(deletes.deletedRowPositions(), numRowsToRead); + } else { + return null; + } + } + + /** + * Build a row id mapping inside a batch, which skips deleted rows. Here is an example of how we delete 2 rows in a + * batch with 8 rows in total. + * [0,1,2,3,4,5,6,7] -- Original status of the row id mapping array + * Position delete 2, 6 + * [0,1,3,4,5,7,-,-] -- After applying position deletes [Set Num records to 6] + * + * @param deletedRowPositions a set of deleted row positions + * @param numRowsToRead the num of rows + * @return the mapping array and the new num of rows in a batch, null if no row is deleted + */ + Pair buildPosDelRowIdMapping(PositionDeleteIndex deletedRowPositions, int numRowsToRead) { + if (deletedRowPositions == null) { + return null; + } + + int[] posDelRowIdMapping = new int[numRowsToRead]; + int originalRowId = 0; + int currentRowId = 0; + while (originalRowId < numRowsToRead) { + if (!deletedRowPositions.deleted(originalRowId + rowStartPosInBatch)) { + posDelRowIdMapping[currentRowId] = originalRowId; + currentRowId++; + } + originalRowId++; + } - return batch.createColumnBatch(arrowColumnVectors); + if (currentRowId == numRowsToRead) { + // there is no delete in this batch + return null; + } else { + return Pair.of(posDelRowIdMapping, currentRowId); + } + } + + int[] initEqDeleteRowIdMapping(int numRowsToRead) { + int[] eqDeleteRowIdMapping = null; + if (hasEqDeletes()) { + eqDeleteRowIdMapping = new int[numRowsToRead]; + for (int i = 0; i < numRowsToRead; i++) { + eqDeleteRowIdMapping[i] = i; + } + } + return eqDeleteRowIdMapping; + } + + /** + * Filter out the equality deleted rows. Here is an example, + * [0,1,2,3,4,5,6,7] -- Original status of the row id mapping array + * Position delete 2, 6 + * [0,1,3,4,5,7,-,-] -- After applying position deletes [Set Num records to 6] + * Equality delete 1 <= x <= 3 + * [0,4,5,7,-,-,-,-] -- After applying equality deletes [Set Num records to 4] + */ + void applyEqDelete() { + Iterator it = columnarBatch.rowIterator(); + int rowId = 0; + int currentRowId = 0; + while (it.hasNext()) { + InternalRow row = it.next(); + if (deletes.eqDeletedRowFilter().test(row)) { + // the row is NOT deleted + // skip deleted rows by pointing to the next undeleted row Id + rowIdMapping[currentRowId] = rowIdMapping[rowId]; + currentRowId++; + } + + rowId++; + } + + columnarBatch.setNumRows(currentRowId); + } } }