Skip to content

Commit ff6f7e5

Browse files
robert3005robert3005
committed
more
1 parent 6d17c91 commit ff6f7e5

6 files changed

Lines changed: 303 additions & 165 deletions

File tree

‎spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/BaseBatchReader.java‎

Lines changed: 93 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -39,13 +39,13 @@
3939
import org.apache.iceberg.io.FileIO;
4040
import org.apache.iceberg.io.InputFile;
4141
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
42-
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
4342
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
4443
import org.apache.iceberg.spark.OrcBatchReadConf;
4544
import org.apache.iceberg.spark.ParquetBatchReadConf;
4645
import org.apache.iceberg.spark.VortexBatchReadConf;
4746
import org.apache.iceberg.spark.data.vectorized.ColumnVectorWithFilter;
4847
import org.apache.iceberg.spark.data.vectorized.ColumnarBatchUtil;
48+
import org.apache.iceberg.spark.data.vectorized.DeletedColumnVector;
4949
import org.apache.iceberg.spark.data.vectorized.UpdatableDeletedColumnVector;
5050
import org.apache.iceberg.types.Types;
5151
import org.apache.iceberg.util.Pair;
@@ -102,9 +102,20 @@ protected CloseableIterable<ColumnarBatch> newBatchIterable(
102102
}
103103

104104
if (readBuilder.supportsPositionDeletes()) {
105-
// Vortex applies position deletes and residual filters natively inside the scan, so the
106-
// post-scan BatchDeleteFilter (which derives row positions from a contiguous _pos column and
107-
// is unsound once rows are filtered out during the scan) is bypassed entirely.
105+
// Vortex applies deletes (and residual filters) natively in the scan, so the post-scan
106+
// BatchDeleteFilter (which derives positions from a contiguous _pos column and is unsound
107+
// once
108+
// rows are filtered out during the scan) is bypassed. Deleted rows are normally dropped via a
109+
// pushed position bitmap; but when the _deleted metadata column is projected they must
110+
// instead
111+
// be retained and flagged, so nothing is pushed and rows are marked from their _pos.
112+
boolean markDeletes =
113+
deleteFilter.requiredSchema().findField(MetadataColumns.IS_DELETED.fieldId()) != null
114+
&& (deleteFilter.hasPosDeletes() || deleteFilter.hasEqDeletes());
115+
if (markDeletes) {
116+
return newDeleteMarkingBatchIterable(
117+
inputFile, readBuilder, start, length, residual, idToConstant, deleteFilter);
118+
}
108119
return newPushdownBatchIterable(
109120
inputFile, readBuilder, start, length, residual, idToConstant, deleteFilter);
110121
}
@@ -127,12 +138,9 @@ protected CloseableIterable<ColumnarBatch> newBatchIterable(
127138
return CloseableIterable.transform(iterable, new BatchDeleteFilter(deleteFilter)::filterBatch);
128139
}
129140

130-
// Reads from a format that applies position deletes (and residual filters) natively in the scan.
131-
// Every delete is turned into file-relative positions and pushed down so deleted rows are never
132-
// materialized: position deletes directly, and equality deletes by a pre-scan that resolves the
133-
// matching rows to positions. Only the expected output columns are projected. The _deleted
134-
// metadata column combined with delete files requires retaining and marking rows, which this
135-
// drop-only path does not do, so that combination is rejected.
141+
// Drops deleted rows by turning every delete into file-relative positions and pushing them into
142+
// the scan so they are never materialized: position deletes directly, and equality deletes via a
143+
// pre-scan that resolves the matching rows to positions. Only the expected columns are projected.
136144
private CloseableIterable<ColumnarBatch> newPushdownBatchIterable(
137145
InputFile inputFile,
138146
ReadBuilder<ColumnarBatch, ?> readBuilder,
@@ -141,13 +149,6 @@ private CloseableIterable<ColumnarBatch> newPushdownBatchIterable(
141149
Expression residual,
142150
Map<Integer, ?> idToConstant,
143151
SparkDeleteFilter deleteFilter) {
144-
boolean isDeletedProjected =
145-
deleteFilter.requiredSchema().findField(MetadataColumns.IS_DELETED.fieldId()) != null;
146-
boolean hasDeletes = deleteFilter.hasPosDeletes() || deleteFilter.hasEqDeletes();
147-
Preconditions.checkArgument(
148-
!(isDeletedProjected && hasDeletes),
149-
"The _deleted metadata column with delete files is not supported for Vortex reads");
150-
151152
PositionDeleteIndex deletePositions =
152153
pushableDeletePositions(inputFile, start, length, residual, idToConstant, deleteFilter);
153154
if (deletePositions.isNotEmpty()) {
@@ -165,6 +166,81 @@ private CloseableIterable<ColumnarBatch> newPushdownBatchIterable(
165166
.build();
166167
}
167168

169+
// Retains all rows and flags deleted ones in the _deleted column instead of dropping them (for
170+
// CDC-style reads that project _deleted). Nothing is pushed into the scan; rows are marked using
171+
// their actual _pos, which stays correct even when a residual filter makes positions
172+
// non-contiguous within a batch.
173+
private CloseableIterable<ColumnarBatch> newDeleteMarkingBatchIterable(
174+
InputFile inputFile,
175+
ReadBuilder<ColumnarBatch, ?> readBuilder,
176+
long start,
177+
long length,
178+
Expression residual,
179+
Map<Integer, ?> idToConstant,
180+
SparkDeleteFilter deleteFilter) {
181+
PositionDeleteIndex deletePositions =
182+
pushableDeletePositions(inputFile, start, length, residual, idToConstant, deleteFilter);
183+
184+
Schema expectedSchema = deleteFilter.expectedSchema();
185+
Schema scanSchema = expectedSchema;
186+
if (expectedSchema.findField(MetadataColumns.ROW_POSITION.fieldId()) == null) {
187+
// _pos is needed to look up the delete state of each row; append it when not already
188+
// projected
189+
// and trim it back off the emitted batch.
190+
List<Types.NestedField> columns = Lists.newArrayList(expectedSchema.columns());
191+
columns.add(MetadataColumns.ROW_POSITION);
192+
scanSchema = new Schema(columns);
193+
}
194+
int rowPositionIndex = scanSchema.columns().indexOf(MetadataColumns.ROW_POSITION);
195+
int isDeletedIndex = scanSchema.columns().indexOf(MetadataColumns.IS_DELETED);
196+
int outputColumnCount = expectedSchema.columns().size();
197+
198+
CloseableIterable<ColumnarBatch> iterable =
199+
readBuilder
200+
.project(scanSchema)
201+
.idToConstant(idToConstant)
202+
.split(start, length)
203+
.filter(residual)
204+
.caseSensitive(caseSensitive())
205+
.reuseContainers()
206+
.withNameMapping(nameMapping())
207+
.build();
208+
209+
return CloseableIterable.transform(
210+
iterable,
211+
batch ->
212+
markDeletedRows(
213+
batch, deletePositions, rowPositionIndex, isDeletedIndex, outputColumnCount));
214+
}
215+
216+
private static ColumnarBatch markDeletedRows(
217+
ColumnarBatch batch,
218+
PositionDeleteIndex deletePositions,
219+
int rowPositionIndex,
220+
int isDeletedIndex,
221+
int outputColumnCount) {
222+
int numRows = batch.numRows();
223+
ColumnVector rowPositions = batch.column(rowPositionIndex);
224+
boolean[] isDeleted = new boolean[numRows];
225+
for (int row = 0; row < numRows; row++) {
226+
isDeleted[row] = deletePositions.isDeleted(rowPositions.getLong(row));
227+
}
228+
229+
DeletedColumnVector deletedColumn = new DeletedColumnVector(Types.BooleanType.get());
230+
deletedColumn.setValue(isDeleted);
231+
232+
// Emit only the expected columns (_pos is dropped when it was appended just for marking) with
233+
// the constant _deleted column replaced by the computed flags.
234+
ColumnVector[] vectors = new ColumnVector[outputColumnCount];
235+
for (int i = 0; i < outputColumnCount; i++) {
236+
vectors[i] = i == isDeletedIndex ? deletedColumn : batch.column(i);
237+
}
238+
239+
ColumnarBatch output = new ColumnarBatch(vectors);
240+
output.setNumRows(numRows);
241+
return output;
242+
}
243+
168244
// Collects all rows to drop as file-relative positions: the position deletes for the file, plus
169245
// the positions of rows matching equality deletes (resolved by a row-oriented pre-scan).
170246
private PositionDeleteIndex pushableDeletePositions(

‎spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkVortexReaderDeletes.java‎

Lines changed: 8 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,12 @@
2727
/**
2828
* Exercises the Spark columnar read path for Vortex tables with deletes.
2929
*
30-
* <p>Vortex applies all deletes (and residual filters) natively in the scan (see {@code
31-
* BaseBatchReader.newPushdownBatchIterable}): position deletes are pushed directly, and equality
32-
* deletes are resolved to positions by a pre-scan and pushed as well. So the position- and
33-
* equality-delete cases inherited from {@link TestSparkReaderDeletes} run unchanged. Only the
34-
* {@code _deleted}-metadata-column-with-delete-files cases are skipped: that requires retaining and
35-
* marking deleted rows, which the drop-only pushdown path does not do.
30+
* <p>Vortex applies deletes (and residual filters) natively in the scan (see {@code
31+
* BaseBatchReader}): when deleted rows are dropped, position deletes are pushed directly and
32+
* equality deletes are resolved to positions by a pre-scan; when the {@code _deleted} column is
33+
* projected, rows are retained and flagged from their {@code _pos} instead. So the position-,
34+
* equality-, and {@code _deleted}-delete cases inherited from {@link TestSparkReaderDeletes} all
35+
* run.
3636
*/
3737
public class TestSparkVortexReaderDeletes extends TestSparkReaderDeletes {
3838

@@ -44,48 +44,18 @@ public static Object[][] parameters() {
4444
};
4545
}
4646

47-
// Deletes are dropped inside the Vortex scan, so they never reach Spark and are not reflected in
48-
// the NumDeletes metric. Disable delete-count assertions for this path.
47+
// Deletes are applied inside the Vortex scan (or marked from _pos), so they never reach Spark and
48+
// are not reflected in the NumDeletes metric. Disable delete-count assertions for this path.
4949
@Override
5050
protected boolean countDeletes() {
5151
return false;
5252
}
5353

54-
private static void skipDeletedColumn() {
55-
Assumptions.abort(
56-
"Vortex applies deletes via native scan pushdown (drop-only); the _deleted metadata column "
57-
+ "with delete files requires retaining and marking rows, which is not supported");
58-
}
59-
6054
@TestTemplate
6155
@Override
6256
public void testReadEqualityDeleteRows() {
6357
// Uses EqualityDeleteRowReader with byte-range task planning; Vortex interprets split ranges as
6458
// row positions, which is a separate limitation from equality-delete read support.
6559
Assumptions.abort("EqualityDeleteRowReader uses byte-range splits, unsupported by Vortex");
6660
}
67-
68-
@TestTemplate
69-
@Override
70-
public void testPosDeletesWithDeletedColumn() {
71-
skipDeletedColumn();
72-
}
73-
74-
@TestTemplate
75-
@Override
76-
public void testEqualityDeleteWithDeletedColumn() {
77-
skipDeletedColumn();
78-
}
79-
80-
@TestTemplate
81-
@Override
82-
public void testMixedPosAndEqDeletesWithDeletedColumn() {
83-
skipDeletedColumn();
84-
}
85-
86-
@TestTemplate
87-
@Override
88-
public void testFilterOnDeletedMetadataColumn() {
89-
skipDeletedColumn();
90-
}
9161
}

‎spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/BaseBatchReader.java‎

Lines changed: 93 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -39,13 +39,13 @@
3939
import org.apache.iceberg.io.FileIO;
4040
import org.apache.iceberg.io.InputFile;
4141
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
42-
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
4342
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
4443
import org.apache.iceberg.spark.OrcBatchReadConf;
4544
import org.apache.iceberg.spark.ParquetBatchReadConf;
4645
import org.apache.iceberg.spark.VortexBatchReadConf;
4746
import org.apache.iceberg.spark.data.vectorized.ColumnVectorWithFilter;
4847
import org.apache.iceberg.spark.data.vectorized.ColumnarBatchUtil;
48+
import org.apache.iceberg.spark.data.vectorized.DeletedColumnVector;
4949
import org.apache.iceberg.spark.data.vectorized.UpdatableDeletedColumnVector;
5050
import org.apache.iceberg.types.Types;
5151
import org.apache.iceberg.util.Pair;
@@ -102,9 +102,20 @@ protected CloseableIterable<ColumnarBatch> newBatchIterable(
102102
}
103103

104104
if (readBuilder.supportsPositionDeletes()) {
105-
// Vortex applies position deletes and residual filters natively inside the scan, so the
106-
// post-scan BatchDeleteFilter (which derives row positions from a contiguous _pos column and
107-
// is unsound once rows are filtered out during the scan) is bypassed entirely.
105+
// Vortex applies deletes (and residual filters) natively in the scan, so the post-scan
106+
// BatchDeleteFilter (which derives positions from a contiguous _pos column and is unsound
107+
// once
108+
// rows are filtered out during the scan) is bypassed. Deleted rows are normally dropped via a
109+
// pushed position bitmap; but when the _deleted metadata column is projected they must
110+
// instead
111+
// be retained and flagged, so nothing is pushed and rows are marked from their _pos.
112+
boolean markDeletes =
113+
deleteFilter.requiredSchema().findField(MetadataColumns.IS_DELETED.fieldId()) != null
114+
&& (deleteFilter.hasPosDeletes() || deleteFilter.hasEqDeletes());
115+
if (markDeletes) {
116+
return newDeleteMarkingBatchIterable(
117+
inputFile, readBuilder, start, length, residual, idToConstant, deleteFilter);
118+
}
108119
return newPushdownBatchIterable(
109120
inputFile, readBuilder, start, length, residual, idToConstant, deleteFilter);
110121
}
@@ -127,12 +138,9 @@ protected CloseableIterable<ColumnarBatch> newBatchIterable(
127138
return CloseableIterable.transform(iterable, new BatchDeleteFilter(deleteFilter)::filterBatch);
128139
}
129140

130-
// Reads from a format that applies position deletes (and residual filters) natively in the scan.
131-
// Every delete is turned into file-relative positions and pushed down so deleted rows are never
132-
// materialized: position deletes directly, and equality deletes by a pre-scan that resolves the
133-
// matching rows to positions. Only the expected output columns are projected. The _deleted
134-
// metadata column combined with delete files requires retaining and marking rows, which this
135-
// drop-only path does not do, so that combination is rejected.
141+
// Drops deleted rows by turning every delete into file-relative positions and pushing them into
142+
// the scan so they are never materialized: position deletes directly, and equality deletes via a
143+
// pre-scan that resolves the matching rows to positions. Only the expected columns are projected.
136144
private CloseableIterable<ColumnarBatch> newPushdownBatchIterable(
137145
InputFile inputFile,
138146
ReadBuilder<ColumnarBatch, ?> readBuilder,
@@ -141,13 +149,6 @@ private CloseableIterable<ColumnarBatch> newPushdownBatchIterable(
141149
Expression residual,
142150
Map<Integer, ?> idToConstant,
143151
SparkDeleteFilter deleteFilter) {
144-
boolean isDeletedProjected =
145-
deleteFilter.requiredSchema().findField(MetadataColumns.IS_DELETED.fieldId()) != null;
146-
boolean hasDeletes = deleteFilter.hasPosDeletes() || deleteFilter.hasEqDeletes();
147-
Preconditions.checkArgument(
148-
!(isDeletedProjected && hasDeletes),
149-
"The _deleted metadata column with delete files is not supported for Vortex reads");
150-
151152
PositionDeleteIndex deletePositions =
152153
pushableDeletePositions(inputFile, start, length, residual, idToConstant, deleteFilter);
153154
if (deletePositions.isNotEmpty()) {
@@ -165,6 +166,81 @@ private CloseableIterable<ColumnarBatch> newPushdownBatchIterable(
165166
.build();
166167
}
167168

169+
// Retains all rows and flags deleted ones in the _deleted column instead of dropping them (for
170+
// CDC-style reads that project _deleted). Nothing is pushed into the scan; rows are marked using
171+
// their actual _pos, which stays correct even when a residual filter makes positions
172+
// non-contiguous within a batch.
173+
private CloseableIterable<ColumnarBatch> newDeleteMarkingBatchIterable(
174+
InputFile inputFile,
175+
ReadBuilder<ColumnarBatch, ?> readBuilder,
176+
long start,
177+
long length,
178+
Expression residual,
179+
Map<Integer, ?> idToConstant,
180+
SparkDeleteFilter deleteFilter) {
181+
PositionDeleteIndex deletePositions =
182+
pushableDeletePositions(inputFile, start, length, residual, idToConstant, deleteFilter);
183+
184+
Schema expectedSchema = deleteFilter.expectedSchema();
185+
Schema scanSchema = expectedSchema;
186+
if (expectedSchema.findField(MetadataColumns.ROW_POSITION.fieldId()) == null) {
187+
// _pos is needed to look up the delete state of each row; append it when not already
188+
// projected
189+
// and trim it back off the emitted batch.
190+
List<Types.NestedField> columns = Lists.newArrayList(expectedSchema.columns());
191+
columns.add(MetadataColumns.ROW_POSITION);
192+
scanSchema = new Schema(columns);
193+
}
194+
int rowPositionIndex = scanSchema.columns().indexOf(MetadataColumns.ROW_POSITION);
195+
int isDeletedIndex = scanSchema.columns().indexOf(MetadataColumns.IS_DELETED);
196+
int outputColumnCount = expectedSchema.columns().size();
197+
198+
CloseableIterable<ColumnarBatch> iterable =
199+
readBuilder
200+
.project(scanSchema)
201+
.idToConstant(idToConstant)
202+
.split(start, length)
203+
.filter(residual)
204+
.caseSensitive(caseSensitive())
205+
.reuseContainers()
206+
.withNameMapping(nameMapping())
207+
.build();
208+
209+
return CloseableIterable.transform(
210+
iterable,
211+
batch ->
212+
markDeletedRows(
213+
batch, deletePositions, rowPositionIndex, isDeletedIndex, outputColumnCount));
214+
}
215+
216+
private static ColumnarBatch markDeletedRows(
217+
ColumnarBatch batch,
218+
PositionDeleteIndex deletePositions,
219+
int rowPositionIndex,
220+
int isDeletedIndex,
221+
int outputColumnCount) {
222+
int numRows = batch.numRows();
223+
ColumnVector rowPositions = batch.column(rowPositionIndex);
224+
boolean[] isDeleted = new boolean[numRows];
225+
for (int row = 0; row < numRows; row++) {
226+
isDeleted[row] = deletePositions.isDeleted(rowPositions.getLong(row));
227+
}
228+
229+
DeletedColumnVector deletedColumn = new DeletedColumnVector(Types.BooleanType.get());
230+
deletedColumn.setValue(isDeleted);
231+
232+
// Emit only the expected columns (_pos is dropped when it was appended just for marking) with
233+
// the constant _deleted column replaced by the computed flags.
234+
ColumnVector[] vectors = new ColumnVector[outputColumnCount];
235+
for (int i = 0; i < outputColumnCount; i++) {
236+
vectors[i] = i == isDeletedIndex ? deletedColumn : batch.column(i);
237+
}
238+
239+
ColumnarBatch output = new ColumnarBatch(vectors);
240+
output.setNumRows(numRows);
241+
return output;
242+
}
243+
168244
// Collects all rows to drop as file-relative positions: the position deletes for the file, plus
169245
// the positions of rows matching equality deletes (resolved by a row-oriented pre-scan).
170246
private PositionDeleteIndex pushableDeletePositions(

0 commit comments

Comments
 (0)