aokolnychyi commented on code in PR #7388: URL: https://github.com/apache/iceberg/pull/7388#discussion_r1180612433
########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/CarryoverRemoveIterator.java: ########## @@ -0,0 +1,100 @@ +/* + * 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; + +import java.util.Iterator; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; + +class CarryoverRemoveIterator extends ChangelogIterator { Review Comment: The logic here is fairly non-trivial. Can we add Javadoc with examples? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java: ########## @@ -59,29 +60,37 @@ * </ul> */ public class ChangelogIterator implements Iterator<Row> { - private static final String DELETE = ChangelogOperation.DELETE.name(); - private static final String INSERT = ChangelogOperation.INSERT.name(); - private static final String UPDATE_BEFORE = ChangelogOperation.UPDATE_BEFORE.name(); - private static final String UPDATE_AFTER = ChangelogOperation.UPDATE_AFTER.name(); + protected static final String DELETE = ChangelogOperation.DELETE.name(); + protected static final String INSERT = ChangelogOperation.INSERT.name(); + protected static final String UPDATE_BEFORE = ChangelogOperation.UPDATE_BEFORE.name(); + protected static final String UPDATE_AFTER = ChangelogOperation.UPDATE_AFTER.name(); private final Iterator<Row> rowIterator; private final int changeTypeIndex; - private final List<Integer> identifierFieldIdx; + private final String[] identifierFields; private final int[] indicesForIdentifySameRow; + private List<Integer> identifierFieldIdx = null; private Row cachedRow = null; - private ChangelogIterator( + protected ChangelogIterator( Iterator<Row> rowIterator, StructType rowType, String[] identifierFields) { this.rowIterator = rowIterator; this.changeTypeIndex = rowType.fieldIndex(MetadataColumns.CHANGE_TYPE.name()); - this.identifierFieldIdx = - Arrays.stream(identifierFields) - .map(column -> rowType.fieldIndex(column.toString())) - .collect(Collectors.toList()); + this.identifierFields = identifierFields; + if (identifierFields != null) { + this.identifierFieldIdx = + Arrays.stream(identifierFields) + .map(column -> rowType.fieldIndex(column.toString())) Review Comment: minor: Redundant `toString()` call? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java: ########## @@ -59,29 +60,37 @@ * </ul> */ public class ChangelogIterator implements Iterator<Row> { Review Comment: Do we have to adjust the Javadoc of this class given that the carryovers are handled in a separate class? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/CarryoverRemoveIterator.java: ########## @@ -0,0 +1,100 @@ +/* + * 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; + +import java.util.Iterator; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; + +class CarryoverRemoveIterator extends ChangelogIterator { + private final Iterator<Row> rowIterator; + + private Row deletedRow = null; + private long deletedRowCount = 0; + private Row nextCachedRow = null; + + CarryoverRemoveIterator(Iterator<Row> rowIterator, StructType rowType) { + super(rowIterator, rowType, null); + this.rowIterator = rowIterator; + } + + @Override + public boolean hasNext() { + if (hasDeleteRow() || nextCachedRow != null) { + return true; + } + return rowIterator.hasNext(); + } + + @Override + public Row next() { + if (popupDeleteRow()) { + deletedRowCount--; + return deletedRow; + } + + Row currentRow = curentRow(); + + if (currentRow.getString(getChangeTypeIndex()).equals(DELETE) && rowIterator.hasNext()) { + // cache the delete row if not done yet + if (!hasDeleteRow()) { + deletedRow = currentRow; + deletedRowCount++; + } + + Row nextRow = rowIterator.next(); + + if (isSameRecord(currentRow, nextRow)) { + if (nextRow.getString(getChangeTypeIndex()).equals(INSERT)) { + deletedRowCount--; + currentRow = null; + } else { + deletedRowCount++; + currentRow = null; + } + } else { + // mark the boundary since the next row is not the same record as the current row + nextCachedRow = nextRow; + currentRow = null; + } + } + + return currentRow; + } + + private boolean popupDeleteRow() { + return (!rowIterator.hasNext() || nextCachedRow != null) && hasDeleteRow(); + } + + private boolean hasDeleteRow() { + return deletedRowCount > 0; + } + + private Row curentRow() { Review Comment: minor: Typo `curent` -> `current`? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java: ########## @@ -121,15 +137,18 @@ public Row next() { Row nextRow = rowIterator.next(); cachedRow = nextRow; - if (isUpdateOrCarryoverRecord(currentRow, nextRow)) { - if (isCarryoverRecord(currentRow, nextRow)) { - // set carry-over rows to null for filtering out later - currentRow = null; - cachedRow = null; - } else { - currentRow = modify(currentRow, changeTypeIndex, UPDATE_BEFORE); - cachedRow = modify(nextRow, changeTypeIndex, UPDATE_AFTER); - } + if (sameLogicalRow(currentRow, nextRow)) { Review Comment: Does this mean we no longer remove carryovers if asked for pre and post images? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/CarryoverRemoveIterator.java: ########## @@ -0,0 +1,100 @@ +/* + * 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; + +import java.util.Iterator; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; + +class CarryoverRemoveIterator extends ChangelogIterator { + private final Iterator<Row> rowIterator; + + private Row deletedRow = null; + private long deletedRowCount = 0; Review Comment: Are there any potential issues with reusing the same object given that we mutate that object in place in pre/post image computation? Or is it safe because there will be an exception if multiple DELETE rows get there? Seems like it is going to work, just checking. ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java: ########## @@ -121,15 +137,18 @@ public Row next() { Row nextRow = rowIterator.next(); cachedRow = nextRow; - if (isUpdateOrCarryoverRecord(currentRow, nextRow)) { - if (isCarryoverRecord(currentRow, nextRow)) { - // set carry-over rows to null for filtering out later - currentRow = null; - cachedRow = null; - } else { - currentRow = modify(currentRow, changeTypeIndex, UPDATE_BEFORE); - cachedRow = modify(nextRow, changeTypeIndex, UPDATE_AFTER); - } + if (sameLogicalRow(currentRow, nextRow)) { Review Comment: Okay, I see that this is called on top of carryover removal now. ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java: ########## @@ -59,29 +60,37 @@ * </ul> */ public class ChangelogIterator implements Iterator<Row> { - private static final String DELETE = ChangelogOperation.DELETE.name(); - private static final String INSERT = ChangelogOperation.INSERT.name(); - private static final String UPDATE_BEFORE = ChangelogOperation.UPDATE_BEFORE.name(); - private static final String UPDATE_AFTER = ChangelogOperation.UPDATE_AFTER.name(); + protected static final String DELETE = ChangelogOperation.DELETE.name(); + protected static final String INSERT = ChangelogOperation.INSERT.name(); + protected static final String UPDATE_BEFORE = ChangelogOperation.UPDATE_BEFORE.name(); + protected static final String UPDATE_AFTER = ChangelogOperation.UPDATE_AFTER.name(); private final Iterator<Row> rowIterator; private final int changeTypeIndex; - private final List<Integer> identifierFieldIdx; + private final String[] identifierFields; private final int[] indicesForIdentifySameRow; + private List<Integer> identifierFieldIdx = null; private Row cachedRow = null; - private ChangelogIterator( + protected ChangelogIterator( Iterator<Row> rowIterator, StructType rowType, String[] identifierFields) { this.rowIterator = rowIterator; this.changeTypeIndex = rowType.fieldIndex(MetadataColumns.CHANGE_TYPE.name()); - this.identifierFieldIdx = - Arrays.stream(identifierFields) - .map(column -> rowType.fieldIndex(column.toString())) - .collect(Collectors.toList()); + this.identifierFields = identifierFields; + if (identifierFields != null) { + this.identifierFieldIdx = + Arrays.stream(identifierFields) + .map(column -> rowType.fieldIndex(column.toString())) + .collect(Collectors.toList()); + } this.indicesForIdentifySameRow = generateIndicesForIdentifySameRow(rowType.size()); } + protected int getChangeTypeIndex() { Review Comment: minor: Just `changeTypeIndex()` as we rarely include `get` prefixes for getters? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/CarryoverRemoveIterator.java: ########## @@ -0,0 +1,100 @@ +/* + * 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; + +import java.util.Iterator; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; + +class CarryoverRemoveIterator extends ChangelogIterator { + private final Iterator<Row> rowIterator; + + private Row deletedRow = null; + private long deletedRowCount = 0; + private Row nextCachedRow = null; + + CarryoverRemoveIterator(Iterator<Row> rowIterator, StructType rowType) { + super(rowIterator, rowType, null); + this.rowIterator = rowIterator; + } + + @Override + public boolean hasNext() { + if (hasDeleteRow() || nextCachedRow != null) { + return true; + } + return rowIterator.hasNext(); + } + + @Override + public Row next() { + if (popupDeleteRow()) { + deletedRowCount--; + return deletedRow; + } + + Row currentRow = curentRow(); + + if (currentRow.getString(getChangeTypeIndex()).equals(DELETE) && rowIterator.hasNext()) { + // cache the delete row if not done yet + if (!hasDeleteRow()) { + deletedRow = currentRow; + deletedRowCount++; + } + + Row nextRow = rowIterator.next(); + + if (isSameRecord(currentRow, nextRow)) { + if (nextRow.getString(getChangeTypeIndex()).equals(INSERT)) { + deletedRowCount--; + currentRow = null; + } else { + deletedRowCount++; + currentRow = null; + } + } else { + // mark the boundary since the next row is not the same record as the current row + nextCachedRow = nextRow; + currentRow = null; + } + } + + return currentRow; + } + + private boolean popupDeleteRow() { Review Comment: Can we add a comment on when this method returns true and what that means for the overall flow? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/CarryoverRemoveIterator.java: ########## @@ -0,0 +1,100 @@ +/* + * 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; + +import java.util.Iterator; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; + +class CarryoverRemoveIterator extends ChangelogIterator { + private final Iterator<Row> rowIterator; + + private Row deletedRow = null; + private long deletedRowCount = 0; + private Row nextCachedRow = null; + + CarryoverRemoveIterator(Iterator<Row> rowIterator, StructType rowType) { + super(rowIterator, rowType, null); + this.rowIterator = rowIterator; + } + + @Override + public boolean hasNext() { + if (hasDeleteRow() || nextCachedRow != null) { + return true; + } + return rowIterator.hasNext(); + } + + @Override + public Row next() { + if (popupDeleteRow()) { + deletedRowCount--; + return deletedRow; + } + + Row currentRow = curentRow(); + + if (currentRow.getString(getChangeTypeIndex()).equals(DELETE) && rowIterator.hasNext()) { + // cache the delete row if not done yet + if (!hasDeleteRow()) { + deletedRow = currentRow; + deletedRowCount++; Review Comment: Why do we only increment the count if this is the first delete? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/CarryoverRemoveIterator.java: ########## @@ -0,0 +1,100 @@ +/* + * 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; + +import java.util.Iterator; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; + +class CarryoverRemoveIterator extends ChangelogIterator { + private final Iterator<Row> rowIterator; + + private Row deletedRow = null; + private long deletedRowCount = 0; + private Row nextCachedRow = null; + + CarryoverRemoveIterator(Iterator<Row> rowIterator, StructType rowType) { + super(rowIterator, rowType, null); + this.rowIterator = rowIterator; + } + + @Override + public boolean hasNext() { + if (hasDeleteRow() || nextCachedRow != null) { + return true; + } + return rowIterator.hasNext(); + } + + @Override + public Row next() { + if (popupDeleteRow()) { + deletedRowCount--; + return deletedRow; + } + + Row currentRow = curentRow(); + + if (currentRow.getString(getChangeTypeIndex()).equals(DELETE) && rowIterator.hasNext()) { + // cache the delete row if not done yet + if (!hasDeleteRow()) { + deletedRow = currentRow; Review Comment: Do we ever reset the delete row? It is not a problem because we rely on counts? ########## spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java: ########## @@ -121,15 +137,18 @@ public Row next() { Row nextRow = rowIterator.next(); cachedRow = nextRow; - if (isUpdateOrCarryoverRecord(currentRow, nextRow)) { - if (isCarryoverRecord(currentRow, nextRow)) { - // set carry-over rows to null for filtering out later - currentRow = null; - cachedRow = null; - } else { - currentRow = modify(currentRow, changeTypeIndex, UPDATE_BEFORE); - cachedRow = modify(nextRow, changeTypeIndex, UPDATE_AFTER); - } + if (sameLogicalRow(currentRow, nextRow)) { + String nextRowChangeType = nextRow.getString(changeTypeIndex); + + Preconditions.checkState( + nextRowChangeType.equals(INSERT), + "The next row should be an INSERT row, but it is %s. That means there are multiple" + + " rows with the same value of identifier fields(%s). Please make sure the rows are unique.", + nextRowChangeType, + identifierFields); Review Comment: Is this array being nicely formatted? Can we add a test to verify the error message? -- 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]
