You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "szehon-ho (via GitHub)" <gi...@apache.org> on 2023/07/13 17:36:15 UTC
[GitHub] [iceberg] szehon-ho opened a new pull request, #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
szehon-ho opened a new pull request, #8059:
URL: https://github.com/apache/iceberg/pull/8059
Fixes: https://github.com/apache/iceberg/issues/8045
This error is from the code that does 'removeDanglingDeletes'.
In this code, for each partition of a group of delete files, I am trying to find 'live' data files so I can find which no longer apply (dangling) and can clean up.
Code: https://github.com/apache/iceberg/blob/master/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java#L122
The issue is, I use the DeleteFile's partition values directly to query data_files table. I thought it would work as the data_files table is using the transformed partition values, just as the DeleteFile partition values should have.
But the partition data of DeleteFile is not the same type as exposed in the Spark metadata table... In particular, its from metadata in Avro form and there's a difference of logical and real Avro types as defined in spec: https://iceberg.apache.org/spec/#avro
The fix is to convert the Avro value into something that can be used for Spark query.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263240187
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -125,10 +128,11 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
IntStream.range(0, fields.size())
.mapToObj(
i -> {
- Class<?> type = fields.get(i).type().typeId().javaClass();
- Object value = partition.get(i, type);
+ Type type = fields.get(i).type();
+ Object value = partition.get(i, type.typeId().javaClass());
+ Object converted = SparkValueConverter.convertToSpark(type, value);
Column col = col("partition." + fields.get(i).name());
- return col.equalTo(value);
+ return col.equalTo(lit(converted));
Review Comment:
yep, done, and added test (manually verified that we gets into partition-pruning ManifestEvaluator isNull block)
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -125,10 +128,11 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
IntStream.range(0, fields.size())
.mapToObj(
i -> {
- Class<?> type = fields.get(i).type().typeId().javaClass();
- Object value = partition.get(i, type);
+ Type type = fields.get(i).type();
+ Object value = partition.get(i, type.typeId().javaClass());
+ Object converted = SparkValueConverter.convertToSpark(type, value);
Column col = col("partition." + fields.get(i).name());
- return col.equalTo(value);
+ return col.equalTo(lit(converted));
Review Comment:
yep, done, and added test (manually verified that we get into partition-pruning ManifestEvaluator isNull block)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263174597
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkValueConverter.java:
##########
@@ -129,4 +130,44 @@ private static Record convert(Types.StructType struct, Row row) {
}
return record;
}
+
+ public static Object convertToSpark(Type type, Object object) {
+ if (object == null) {
+ return null;
+ }
+
+ switch (type.typeId()) {
+ case STRUCT:
+ case LIST:
+ case MAP:
+ return new UnsupportedOperationException("Complex types currently not supported");
+ case DATE:
+ // if spark.sql.datetime.java8API.enabled is set to true, java.time.LocalDate
Review Comment:
How relevant is this comment?
We needed it for converting to Iceberg as it may be either `Date` or `LocalDate`.
It seems redundant here as we are converting number of days to `LocalDate`.
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -125,10 +128,11 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
IntStream.range(0, fields.size())
.mapToObj(
i -> {
- Class<?> type = fields.get(i).type().typeId().javaClass();
- Object value = partition.get(i, type);
+ Type type = fields.get(i).type();
+ Object value = partition.get(i, type.typeId().javaClass());
+ Object converted = SparkValueConverter.convertToSpark(type, value);
Review Comment:
Minor: `converted` -> `convertedValue`?
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkValueConverter.java:
##########
@@ -129,4 +130,44 @@ private static Record convert(Types.StructType struct, Row row) {
}
return record;
}
+
+ public static Object convertToSpark(Type type, Object object) {
+ if (object == null) {
+ return null;
+ }
+
+ switch (type.typeId()) {
+ case STRUCT:
+ case LIST:
+ case MAP:
+ return new UnsupportedOperationException("Complex types currently not supported");
+ case DATE:
+ // if spark.sql.datetime.java8API.enabled is set to true, java.time.LocalDate
+ // for Spark SQL DATE type otherwise java.sql.Date is returned.
+ return DateTimeUtils.daysToLocalDate((int) object);
+ case TIMESTAMP:
+ Types.TimestampType ts = (Types.TimestampType) type.asPrimitiveType();
+ // This if/else can be removed once https://github.com/apache/spark/pull/41238 is in
Review Comment:
None of these comments apply here. The implementation seems valid, though.
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -125,10 +128,11 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
IntStream.range(0, fields.size())
.mapToObj(
i -> {
- Class<?> type = fields.get(i).type().typeId().javaClass();
- Object value = partition.get(i, type);
+ Type type = fields.get(i).type();
+ Object value = partition.get(i, type.typeId().javaClass());
+ Object converted = SparkValueConverter.convertToSpark(type, value);
Column col = col("partition." + fields.get(i).name());
- return col.equalTo(value);
+ return col.equalTo(lit(converted));
Review Comment:
Does this work correctly when the partition value is null? Do we have to use `eqNullSafe`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1262972615
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -138,4 +142,30 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
return SparkTableUtil.loadMetadataTable(spark, table(), MetadataTableType.DATA_FILES);
}
}
+
+ /**
+ * If necessary, convert the partition data (from avro) to its logical type, to use in the spark
+ * queries. See https://iceberg.apache.org/spec/#avro.
+ *
+ * @param value partition object
+ * @param typeId iceberg type
+ * @return object converted to its logical type
+ */
+ private Object convertPartitionValue(Object value, Type.TypeID typeId) {
+ switch (typeId) {
Review Comment:
I understand there is a type mismatch but don't we have to convert internal Iceberg values for a given type to a Spark value? Why Avro?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1262884053
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFiles.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFiles extends SparkExtensionsTestBase {
Review Comment:
Notes:
1. I can probably unite the other TestRewritePositionDeleteFilesAction into this one. That one was without sql-extensions, and I find it a lot easier to populate tables with sql-support, instead of making new java class for each type of partition I am trying to test.
2. This tests all the partition types that may error in : https://iceberg.apache.org/spec/#avro, in regards to removeDanglingDeletes query. Note some of these types do not exist in spark (UUID, time)
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFiles.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFiles extends SparkExtensionsTestBase {
Review Comment:
Notes:
1. I can probably unite the other TestRewritePositionDeleteFilesAction into this one later. That one was without sql-extensions, and I find it a lot easier to populate tables with sql-support, instead of making new java class for each type of partition I am trying to test.
2. This tests all the partition types that may error in : https://iceberg.apache.org/spec/#avro, in regards to removeDanglingDeletes query. Note some of these types do not exist in spark (UUID, time)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1262880694
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFilesAction.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFilesAction extends SparkExtensionsTestBase {
Review Comment:
Notes:
1. I can probably unite the other TestRewritePositionDeleteFilesAction into this one. That one was without sql-extensions, and I find it a lot easier to populate tables with sql-support, instead of making new java class for each type of partition I am trying to test.
2. This tests all the partition types that may error in : https://iceberg.apache.org/spec/#avro, in regards to removeDanglingDeletes query. Note some of these types do not exist in spark (UUID, time)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1262884053
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFiles.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFiles extends SparkExtensionsTestBase {
Review Comment:
Notes:
1. I can probably unite the other TestRewritePositionDeleteFilesAction into this one later. That one was not in project with sql-extensions, and I find it a lot easier to populate tables with sql-support, instead of making new java class for each type of partition I am trying to test.
2. This tests all the partition types that may error in : https://iceberg.apache.org/spec/#avro, in regards to removeDanglingDeletes query. Note some of these types do not exist in spark (UUID, time)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] nastra commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263692477
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFiles.java:
##########
@@ -0,0 +1,417 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFiles extends SparkExtensionsTestBase {
+
+ private static final Map<String, String> CATALOG_PROPS =
+ ImmutableMap.of(
+ "type", "hive",
+ "default-namespace", "default",
+ "cache-enabled", "false");
+
+ private static final int NUM_DATA_FILES = 5;
+ private static final int ROWS_PER_DATA_FILE = 100;
+ private static final int DELETE_FILES_PER_PARTITION = 2;
+ private static final int DELETE_FILE_SIZE = 10;
+
+ @Parameterized.Parameters(
+ name = "formatVersion = {0}, catalogName = {1}, implementation = {2}, config = {3}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.HIVE.catalogName(),
+ SparkCatalogConfig.HIVE.implementation(),
+ CATALOG_PROPS
+ }
+ };
+ }
+
+ @Rule public TemporaryFolder temp = new TemporaryFolder();
+
+ public TestRewritePositionDeleteFiles(
+ String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void cleanup() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testDatePartition() throws Exception {
+ createTable("date");
+ Date baseDate = Date.valueOf("2023-01-01");
+ insertData(i -> Date.valueOf(baseDate.toLocalDate().plusDays(i)));
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testBooleanPartition() throws Exception {
+ createTable("boolean");
+ insertData(i -> i % 2 == 0, 2);
+ testDanglingDelete(2);
+ }
+
+ @Test
+ public void testTimestampPartition() throws Exception {
+ createTable("timestamp");
+ Timestamp baseTimestamp = Timestamp.valueOf("2023-01-01 15:30:00");
+ insertData(i -> Timestamp.valueOf(baseTimestamp.toLocalDateTime().plusDays(i)));
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testTimestampNtz() throws Exception {
+ createTable("timestamp_ntz");
+ LocalDateTime baseTimestamp = Timestamp.valueOf("2023-01-01 15:30:00").toLocalDateTime();
+ insertData(baseTimestamp::plusDays);
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testBytePartition() throws Exception {
+ createTable("byte");
+ insertData(i -> i);
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testDecimalPartition() throws Exception {
+ createTable("decimal(18, 10)");
+ BigDecimal baseDecimal = new BigDecimal("1.0");
+ insertData(i -> baseDecimal.add(new BigDecimal(i)));
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testBinaryPartition() throws Exception {
+ createTable("binary");
+ insertData(i -> java.nio.ByteBuffer.allocate(4).putInt(i).array());
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testCharPartition() throws Exception {
+ createTable("char(10)");
+ insertData(Object::toString);
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testVarcharPartition() throws Exception {
+ createTable("varchar(10)");
+ insertData(Object::toString);
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testIntPartition() throws Exception {
+ createTable("int");
+ insertData(i -> i);
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testDaysPartitionTransform() throws Exception {
+ createTable("timestamp", "days(partition_col)");
+ Timestamp baseTimestamp = Timestamp.valueOf("2023-01-01 15:30:00");
+ insertData(i -> Timestamp.valueOf(baseTimestamp.toLocalDateTime().plusDays(i)));
+ testDanglingDelete();
+ }
+
+ @Test
+ public void testNullTransform() throws Exception {
+ createTable("int");
+ insertData(i -> i == 0 ? null : 1, 2);
+ testDanglingDelete(2);
+ }
+
+ private <T> void testDanglingDelete() throws Exception {
+ testDanglingDelete(NUM_DATA_FILES);
+ }
+
+ private <T> void testDanglingDelete(int numDataFiles) throws Exception {
+ Table table = Spark3Util.loadIcebergTable(spark, tableName);
+
+ List<DataFile> dataFiles = dataFiles(table);
+ Assert.assertEquals(numDataFiles, dataFiles.size());
+
+ SparkActions.get(spark)
+ .rewriteDataFiles(table)
+ .option(SizeBasedFileRewriter.REWRITE_ALL, "true")
+ .execute();
+
+ // write dangling delete files for 'old data files'
+ writePosDeletesForFiles(table, dataFiles);
+ List<DeleteFile> deleteFiles = deleteFiles(table);
+ Assert.assertEquals(numDataFiles * DELETE_FILES_PER_PARTITION, deleteFiles.size());
+
+ List<Object[]> expectedRecords = records(tableName);
+
+ Result result =
+ SparkActions.get(spark)
+ .rewritePositionDeletes(table)
+ .option(SizeBasedFileRewriter.REWRITE_ALL, "true")
+ .execute();
+
+ List<DeleteFile> newDeleteFiles = deleteFiles(table);
+ Assert.assertEquals("Should have removed all dangling delete files", 0, newDeleteFiles.size());
+ checkResult(result, deleteFiles, Lists.newArrayList(), numDataFiles);
+
+ List<Object[]> actualRecords = records(tableName);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ }
+
+ private void createTable(String partitionType) {
+ createTable(partitionType, "partition_col");
+ }
+
+ private void createTable(String partitionType, String partitionCol) {
+ sql(
+ "CREATE TABLE %s (id long, partition_col %s, c1 string, c2 string) "
+ + "USING iceberg "
+ + "PARTITIONED BY (%s) "
+ + "TBLPROPERTIES('format-version'='2')",
+ tableName, partitionType, partitionCol);
+ }
+
+ private <T> void insertData(Function<Integer, ?> partitionValueFunction) throws Exception {
+ insertData(partitionValueFunction, NUM_DATA_FILES);
+ }
+
+ private <T> void insertData(Function<Integer, ?> partitionValue, int numDataFiles)
+ throws Exception {
+ for (int i = 0; i < numDataFiles; i++) {
+ Dataset<Row> df =
+ spark
+ .range(0, ROWS_PER_DATA_FILE)
+ .withColumn("partition_col", lit(partitionValue.apply(i)))
+ .withColumn("c1", expr("CAST(id AS STRING)"))
+ .withColumn("c2", expr("CAST(id AS STRING)"));
+ appendAsFile(df);
+ }
+ }
+
+ private void appendAsFile(Dataset<Row> df) throws Exception {
+ // ensure the schema is precise
+ StructType sparkSchema = spark.table(tableName).schema();
+ spark.createDataFrame(df.rdd(), sparkSchema).coalesce(1).writeTo(tableName).append();
+ }
+
+ private void writePosDeletesForFiles(Table table, List<DataFile> files) throws IOException {
+
+ Map<StructLike, List<DataFile>> filesByPartition =
+ files.stream().collect(Collectors.groupingBy(ContentFile::partition));
+ List<DeleteFile> deleteFiles =
+ Lists.newArrayListWithCapacity(DELETE_FILES_PER_PARTITION * filesByPartition.size());
+
+ for (Map.Entry<StructLike, List<DataFile>> filesByPartitionEntry :
+ filesByPartition.entrySet()) {
+
+ StructLike partition = filesByPartitionEntry.getKey();
+ List<DataFile> partitionFiles = filesByPartitionEntry.getValue();
+
+ int deletesForPartition = partitionFiles.size() * DELETE_FILE_SIZE;
+ Assert.assertEquals(
+ "Number of delete files per partition should be "
+ + "evenly divisible by requested deletes per data file times number of data files in this partition",
+ 0,
+ deletesForPartition % DELETE_FILE_SIZE);
+ int deleteFileSize = deletesForPartition / DELETE_FILES_PER_PARTITION;
+
+ int counter = 0;
+ List<Pair<CharSequence, Long>> deletes = Lists.newArrayList();
+ for (DataFile partitionFile : partitionFiles) {
+ for (int deletePos = 0; deletePos < DELETE_FILE_SIZE; deletePos++) {
+ deletes.add(Pair.of(partitionFile.path(), (long) deletePos));
+ counter++;
+ if (counter == deleteFileSize) {
+ // Dump to file and reset variables
+ OutputFile output = Files.localOutput(temp.newFile());
+ deleteFiles.add(writeDeleteFile(table, output, partition, deletes));
+ counter = 0;
+ deletes.clear();
+ }
+ }
+ }
+ }
+
+ RowDelta rowDelta = table.newRowDelta();
+ deleteFiles.forEach(rowDelta::addDeletes);
+ rowDelta.commit();
+ }
+
+ private DeleteFile writeDeleteFile(
+ Table table, OutputFile out, StructLike partition, List<Pair<CharSequence, Long>> deletes)
+ throws IOException {
+ FileFormat format = defaultFormat(table.properties());
+ FileAppenderFactory<Record> factory = new GenericAppenderFactory(table.schema(), table.spec());
+
+ PositionDeleteWriter<Record> writer =
+ factory.newPosDeleteWriter(encrypt(out), format, partition);
+ PositionDelete<Record> posDelete = PositionDelete.create();
+ try (Closeable toClose = writer) {
+ for (Pair<CharSequence, Long> delete : deletes) {
+ writer.write(posDelete.set(delete.first(), delete.second(), null));
+ }
+ }
+
+ return writer.toDeleteFile();
+ }
+
+ private static EncryptedOutputFile encrypt(OutputFile out) {
+ return EncryptedFiles.encryptedOutput(out, EncryptionKeyMetadata.EMPTY);
+ }
+
+ private static FileFormat defaultFormat(Map<String, String> properties) {
+ String formatString = properties.getOrDefault(DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT);
+ return FileFormat.fromString(formatString);
+ }
+
+ private List<Object[]> records(String table) {
+ return rowsToJava(
+ spark.read().format("iceberg").load(table).sort("partition_col", "id").collectAsList());
+ }
+
+ private long size(List<DeleteFile> deleteFiles) {
+ return deleteFiles.stream().mapToLong(DeleteFile::fileSizeInBytes).sum();
+ }
+
+ private List<DataFile> dataFiles(Table table) {
+ CloseableIterable<FileScanTask> tasks = table.newScan().includeColumnStats().planFiles();
+ return Lists.newArrayList(CloseableIterable.transform(tasks, FileScanTask::file));
+ }
+
+ private List<DeleteFile> deleteFiles(Table table) {
+ Table deletesTable =
+ MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES);
+ CloseableIterable<ScanTask> tasks = deletesTable.newBatchScan().planFiles();
+ return Lists.newArrayList(
+ CloseableIterable.transform(tasks, t -> ((PositionDeletesScanTask) t).file()));
+ }
+
+ private void checkResult(
+ Result result,
+ List<DeleteFile> rewrittenDeletes,
+ List<DeleteFile> newDeletes,
+ int expectedGroups) {
+ Assert.assertEquals(
Review Comment:
Iceberg is currently moving away from Junit4 to Junit5 and in the same step is updating assertions to AssertJ. It would be great to adjust all of these assertions here to AssertJ, as otherwise it'll add additional work later on
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263240187
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -125,10 +128,11 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
IntStream.range(0, fields.size())
.mapToObj(
i -> {
- Class<?> type = fields.get(i).type().typeId().javaClass();
- Object value = partition.get(i, type);
+ Type type = fields.get(i).type();
+ Object value = partition.get(i, type.typeId().javaClass());
+ Object converted = SparkValueConverter.convertToSpark(type, value);
Column col = col("partition." + fields.get(i).name());
- return col.equalTo(value);
+ return col.equalTo(lit(converted));
Review Comment:
yep, done, and added test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#issuecomment-1634809059
@aokolnychyi added a test with partition transform (as discussed offline )
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho merged pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho merged PR #8059:
URL: https://github.com/apache/iceberg/pull/8059
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263139408
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFiles.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFiles extends SparkExtensionsTestBase {
Review Comment:
Ignore point 2, as after discussion with @aokolnychyi its not really avro but just Iceberg type in general that need conversion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263240187
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -125,10 +128,11 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
IntStream.range(0, fields.size())
.mapToObj(
i -> {
- Class<?> type = fields.get(i).type().typeId().javaClass();
- Object value = partition.get(i, type);
+ Type type = fields.get(i).type();
+ Object value = partition.get(i, type.typeId().javaClass());
+ Object converted = SparkValueConverter.convertToSpark(type, value);
Column col = col("partition." + fields.get(i).name());
- return col.equalTo(value);
+ return col.equalTo(lit(converted));
Review Comment:
yep, done, and added test (manually verified that it gets into isNull block)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#issuecomment-1636161627
Sure let me do this in follow up pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1262905937
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -138,4 +142,30 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
return SparkTableUtil.loadMetadataTable(spark, table(), MetadataTableType.DATA_FILES);
}
}
+
+ /**
+ * If necessary, convert the partition data (from avro) to its logical type, to use in the spark
+ * queries. See https://iceberg.apache.org/spec/#avro.
+ *
+ * @param value partition object
+ * @param typeId iceberg type
+ * @return object converted to its logical type
+ */
+ private Object convertPartitionValue(Object value, Type.TypeID typeId) {
+ switch (typeId) {
Review Comment:
The expected java class is from: https://github.com/apache/iceberg/blob/master/api/src/main/java/org/apache/iceberg/types/Type.java#L31
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263138940
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -138,4 +142,30 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
return SparkTableUtil.loadMetadataTable(spark, table(), MetadataTableType.DATA_FILES);
}
}
+
+ /**
+ * If necessary, convert the partition data (from avro) to its logical type, to use in the spark
+ * queries. See https://iceberg.apache.org/spec/#avro.
+ *
+ * @param value partition object
+ * @param typeId iceberg type
+ * @return object converted to its logical type
+ */
+ private Object convertPartitionValue(Object value, Type.TypeID typeId) {
+ switch (typeId) {
Review Comment:
Yea, makes sense, its not really to do with Avro, but indeed we need an Iceberg => Spark direction, just as we have a Spark => Iceberg direction in SparkValueConverter. Added that in latest commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1262880694
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFilesAction.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFilesAction extends SparkExtensionsTestBase {
Review Comment:
Notes:
1. I can probably unite the other TestRewritePositionDeleteFilesAction into this one. That one was without sql-extensions, and I find it a lot easier to populate tables with sql-support, instead of making new java class for each type of partition I am trying to test.
2. This tests all the partition types that may error in : https://iceberg.apache.org/spec/#avro, in regards to removeDanglingDeletes query. Note some of these types do not exist in spark (UUID, time)
##########
spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewritePositionDeleteFiles.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * 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.extensions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.PositionDeletesScanTask;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.FileGroupRewriteResult;
+import org.apache.iceberg.actions.RewritePositionDeleteFiles.Result;
+import org.apache.iceberg.actions.SizeBasedFileRewriter;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedFiles;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.actions.SparkActions;
+import org.apache.iceberg.util.Pair;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runners.Parameterized;
+
+public class TestRewritePositionDeleteFiles extends SparkExtensionsTestBase {
Review Comment:
Notes:
I can probably unite the other TestRewritePositionDeleteFilesAction into this one. That one was without sql-extensions, and I find it a lot easier to populate tables with sql-support, instead of making new java class for each type of partition I am trying to test.
This tests all the partition types that may error in : https://iceberg.apache.org/spec/#avro, in regards to removeDanglingDeletes query. Note some of these types do not exist in spark (UUID, time)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #8059: Spark 3.4: Fix rewrite_position_deletes for certain partition types
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #8059:
URL: https://github.com/apache/iceberg/pull/8059#discussion_r1263028888
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/SparkBinPackPositionDeletesRewriter.java:
##########
@@ -138,4 +142,30 @@ private Dataset<Row> dataFiles(Types.StructType partitionType, StructLike partit
return SparkTableUtil.loadMetadataTable(spark, table(), MetadataTableType.DATA_FILES);
}
}
+
+ /**
+ * If necessary, convert the partition data (from avro) to its logical type, to use in the spark
+ * queries. See https://iceberg.apache.org/spec/#avro.
+ *
+ * @param value partition object
+ * @param typeId iceberg type
+ * @return object converted to its logical type
+ */
+ private Object convertPartitionValue(Object value, Type.TypeID typeId) {
+ switch (typeId) {
Review Comment:
We are getting back Iceberg literals we need to come up with Spark literals. We have `PartitionData` under the hood that is populated by Avro but that should not matter at all. We need to build an reverser conversion similar to what we already have in `SparkValueConverter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org