aokolnychyi commented on code in PR #9437:
URL: https://github.com/apache/iceberg/pull/9437#discussion_r1453851019
##########
api/src/main/java/org/apache/iceberg/actions/ActionsProvider.java:
##########
@@ -70,4 +70,10 @@ default RewritePositionDeleteFiles
rewritePositionDeletes(Table table) {
throw new UnsupportedOperationException(
this.getClass().getName() + " does not implement
rewritePositionDeletes");
}
+
+ /** Instantiates an action to compute partition statistics and register it
to table metadata */
Review Comment:
Minor: Missing dot at the end of the sentence?
##########
api/src/main/java/org/apache/iceberg/actions/ComputePartitionStats.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.actions;
+
+import org.apache.iceberg.PartitionStatisticsFile;
+
+/**
+ * An action to compute partition stats for the latest snapshot and registers
it to the
+ * TableMetadata file
+ */
+public interface ComputePartitionStats
Review Comment:
Question: Do we have to support branches/tags/snapshot IDs? It is not
required in the initial version but I guess it makes sense in general?
##########
api/src/main/java/org/apache/iceberg/actions/ComputePartitionStats.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.actions;
+
+import org.apache.iceberg.PartitionStatisticsFile;
+
+/**
+ * An action to compute partition stats for the latest snapshot and registers
it to the
Review Comment:
What about this? I am not sure I'd mention `TableMetadata` class and
necessarily limit ourselves to the latest snapshot, we may support branches as
well in the future.
```
An action to compute and register partition stats.
```
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/ComputePartitionStatsSparkAction.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.actions;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ImmutableGenericPartitionStatisticsFile;
+import org.apache.iceberg.PartitionData;
+import org.apache.iceberg.PartitionEntry;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionStatisticsFile;
+import org.apache.iceberg.Partitioning;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.ComputePartitionStats;
+import org.apache.iceberg.actions.ImmutableComputePartitionStats;
+import org.apache.iceberg.data.PartitionStatsUtil;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.JobGroupInfo;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PartitionUtil;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An implementation of {@link ComputePartitionStats} that computes and
registers the partition
+ * stats to table metadata
+ */
+public class ComputePartitionStatsSparkAction
+ extends BaseSparkAction<ComputePartitionStatsSparkAction> implements
ComputePartitionStats {
+ private static final Logger LOG =
LoggerFactory.getLogger(ComputePartitionStatsSparkAction.class);
+ private final Table table;
+
+ ComputePartitionStatsSparkAction(SparkSession spark, Table table) {
+ super(spark);
+ this.table = table;
+ }
+
+ @Override
+ protected ComputePartitionStatsSparkAction self() {
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ String jobDesc = String.format("Computing partition stats for the table
%s", table.name());
+ JobGroupInfo info = newJobGroupInfo("COMPUTE-PARTITION-STATS", jobDesc);
+ return withJobGroupInfo(info, this::doExecute);
+ }
+
+ private Result doExecute() {
+ long currentSnapshotId = table.currentSnapshot().snapshotId();
Review Comment:
Hm, won't this generate NPE as the current snapshot would be null if the
table is empty? Do we have a test for this? Also, shouldn't we return a valid
result object with null output file rather than null result?
##########
api/src/main/java/org/apache/iceberg/actions/ComputePartitionStats.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.actions;
+
+import org.apache.iceberg.PartitionStatisticsFile;
+
+/**
+ * An action to compute partition stats for the latest snapshot and registers
it to the
+ * TableMetadata file
+ */
+public interface ComputePartitionStats
+ extends Action<ComputePartitionStats, ComputePartitionStats.Result> {
+
+ /** The action result that contains a summary of the execution. */
+ interface Result {
+ /** Returns the output file which is registered to the table metadata. */
+ PartitionStatisticsFile outputFile();
Review Comment:
Can this value be null if the action is invoked on an empty table? If so,
shall we document that?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -150,6 +154,21 @@ protected Dataset<FileInfo> contentFileDS(Table table,
Set<Long> snapshotIds) {
Broadcast<Table> tableBroadcast =
sparkContext.broadcast(serializableTable);
int numShufflePartitions =
spark.sessionState().conf().numShufflePartitions();
+ return manifestBeanDS(table, snapshotIds, numShufflePartitions)
+ .flatMap(new ReadManifest(tableBroadcast), FileInfo.ENCODER);
+ }
+
+ protected Dataset<PartitionEntryBean> partitionEntryDS(Table table) {
+ Table serializableTable = SerializableTableWithSize.copyOf(table);
+ Broadcast<Table> tableBroadcast =
sparkContext.broadcast(serializableTable);
+ int numShufflePartitions =
spark.sessionState().conf().numShufflePartitions();
+
+ return manifestBeanDS(table, null, numShufflePartitions)
Review Comment:
Is it actually correct? This code would go via ALL_MANIFESTS table.
Shouldn't we only look for manifests in a particular snapshot for which we
compute the stats?
In general, I am not sure how I feel about the current approach. It seems
the only benefit of using the Spark resources here is to parallelize the read
of manifests. However, we bring the entire content of the dataset to the driver
and then try to merge it. This operation will be costly and may diminish the
benefits of using the cluster resources for reading in the first place. Even
though it is an action, it does not necessarily need to utilize the cluster.
I'd probably modify the distributed algorithm and also compare it against a
local implementation.
**Potential Distributed Algorithm**
- Load `entries` metadata table for the snapshot for which we compute stats
as `Dataset<Row>`. I believe we can use `loadMetadataTable` from
`BaseSparkAction` for that.
- Distribute these records by partition using hash distribution to co-locate
entries for the same partition next to each other. Still use plain `Row` for
this.
- Have either a SQL expression (preferable) or closure (still OK but will
require extra serialization) to squash these records and only have one record
per partition as output.
- Collect the squashed results to the driver.
- Write the records into a partition stats file.
Compared to the current solution, we will not only parallelize the reading
phase but also the reduction of entries. This will also lower the transfer and
serialization costs, potentially making the distributed approach worth the
extra complexity (has to be proved).
**Potential Local Algorithm**
In any case, I'd also compare a purely local solution with a thread pool.
- Open all manifests for the given snapshot concurrently.
- Either use a common concurrent hash map to hold reduced values and update
it in each thread OR reduce each manifest concurrently and then merge results
across manifests in a single thread.
I'd start by creating an efficient local implementation and testing it with
1 and 10 million files. It should be fairly simple now by using
`FileGenerationUtil`. We have plenty of benchmarks that leverage that.
Afterwards, we can see if a distributed approach is necessary.
--
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-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
