You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "gaborkaszab (via GitHub)" <gi...@apache.org> on 2025/11/05 11:30:49 UTC
[PR] API, Core, Spark: Scan API for partition stats [iceberg]
gaborkaszab opened a new pull request, #14508:
URL: https://github.com/apache/iceberg/pull/14508
(no comment)
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2498252381
##########
api/src/main/java/org/apache/iceberg/PartitionStatisticsScan.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+
+public interface PartitionStatisticsScan {
+
+ /**
+ * Create a new scan from this scan's configuration that will use the given snapshot by ID.
+ *
+ * @param snapshotId a snapshot ID
+ * @return a new scan based on this with the given snapshot ID
+ * @throws IllegalArgumentException if the snapshot cannot be found
+ */
+ PartitionStatisticsScan useSnapshot(long snapshotId);
+
+ /**
+ * Create a new scan from the results of this, where partitions are filtered by the {@link
+ * Expression}.
+ *
+ * @param filter a filter expression
+ * @return a new scan based on this with results filtered by the expression
+ */
+ PartitionStatisticsScan filter(Expression filter);
+
+ /**
+ * Create a new scan from this with the schema as its projection.
Review Comment:
Maybe describe what will happen with the `PartitionStatistics` attributes which are not part of the schema.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2534718692
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatisticsScan.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Optional;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatisticsScan implements PartitionStatisticsScan {
+
+ private final Table table;
+ private Long snapshotId;
+
+ public BasePartitionStatisticsScan(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ public PartitionStatisticsScan useSnapshot(long newSnapshotId) {
+ Preconditions.checkArgument(
+ table.snapshot(newSnapshotId) != null, "Cannot find snapshot with ID %s", newSnapshotId);
+
+ this.snapshotId = newSnapshotId;
+ return this;
+ }
+
+ @Override
+ public PartitionStatisticsScan filter(Expression newFilter) {
+ throw new UnsupportedOperationException("Filtering is not supported");
+ }
+
+ @Override
+ public PartitionStatisticsScan project(Schema newSchema) {
+ throw new UnsupportedOperationException("Projection is not supported");
+ }
+
+ @Override
+ public CloseableIterable<PartitionStatistics> scan() {
Review Comment:
I added a test suite for this too. Introduced a base class to share functionality with TestBasePartitionStatsHandler
--
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
Re: [PR] PoC: API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#issuecomment-3703094078
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2510907006
##########
api/src/main/java/org/apache/iceberg/PartitionStatisticsScan.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+
+public interface PartitionStatisticsScan {
+
+ /**
+ * Create a new scan from this scan's configuration that will use the given snapshot by ID.
+ *
+ * @param snapshotId a snapshot ID
+ * @return a new scan based on this with the given snapshot ID
+ * @throws IllegalArgumentException if the snapshot cannot be found
+ */
+ PartitionStatisticsScan useSnapshot(long snapshotId);
+
+ /**
+ * Create a new scan from the results of this, where partitions are filtered by the {@link
+ * Expression}.
+ *
+ * @param filter a filter expression
+ * @return a new scan based on this with results filtered by the expression
+ */
+ PartitionStatisticsScan filter(Expression filter);
+
+ /**
+ * Create a new scan from this with the schema as its projection.
+ *
+ * @param schema a projection schema
Review Comment:
You're right. Let me add this to the possible follow-up steps in my comment at the top
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2542576597
##########
api/src/main/java/org/apache/iceberg/PartitionStatisticsScan.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+
+public interface PartitionStatisticsScan {
Review Comment:
done
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -275,7 +275,9 @@ static PartitionStatisticsFile writePartitionStatsFile(
*
* @param schema The {@link Schema} of the partition statistics file.
* @param inputFile An {@link InputFile} pointing to the partition stats file.
+ * @deprecated will be removed in 1.12.0, use {@link PartitionStatisticsScan} instead
*/
+ @Deprecated
public static CloseableIterable<PartitionStats> readPartitionStatsFile(
Review Comment:
Marked the tests exercising the old functionality as deprecated simply for visibility.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2498293303
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -349,7 +350,7 @@ private static Collection<PartitionStats> computeAndMergeStatsIncremental(
Pair.of(key.first(), partitionDataToRecord((PartitionData) key.second())),
value,
(existingEntry, newEntry) -> {
- existingEntry.appendStats(newEntry);
+ ((PartitionStats) existingEntry).appendStats(newEntry);
Review Comment:
I would prefer to keep the interface clean
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2537341524
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -275,7 +275,9 @@ static PartitionStatisticsFile writePartitionStatsFile(
*
* @param schema The {@link Schema} of the partition statistics file.
* @param inputFile An {@link InputFile} pointing to the partition stats file.
+ * @deprecated will be removed in 1.12.0, use {@link PartitionStatisticsScan} instead
*/
+ @Deprecated
public static CloseableIterable<PartitionStats> readPartitionStatsFile(
Review Comment:
FYI, I added the relevant tests back that exercise this deprecated function.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2523133744
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatistics.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * 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;
+
+import org.apache.iceberg.avro.SupportsIndexProjection;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatistics extends SupportsIndexProjection
+ implements PartitionStatistics {
+
+ private StructLike partition;
+ private Integer specId;
+ private Long dataRecordCount;
+ private Integer dataFileCount;
+ private Long totalDataFileSizeInBytes;
+ private Long positionDeleteRecordCount;
+ private Integer positionDeleteFileCount;
+ private Long equalityDeleteRecordCount;
+ private Integer equalityDeleteFileCount;
+ private Long totalRecordCount; // null by default
+ private Long lastUpdatedAt; // null by default
+ private Long lastUpdatedSnapshotId; // null by default
Review Comment:
It is hard to understand the comment.
Are the others not null by default?
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatisticsScan.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Optional;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatisticsScan implements PartitionStatisticsScan {
+
+ private final Table table;
+ private Long snapshotId;
+
+ public BasePartitionStatisticsScan(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ public PartitionStatisticsScan useSnapshot(long newSnapshotId) {
+ Preconditions.checkArgument(
+ table.snapshot(newSnapshotId) != null, "Cannot find snapshot with ID %s", newSnapshotId);
+
+ this.snapshotId = newSnapshotId;
+ return this;
+ }
+
+ @Override
+ public PartitionStatisticsScan filter(Expression newFilter) {
+ throw new UnsupportedOperationException("Filtering is not supported");
+ }
+
+ @Override
+ public PartitionStatisticsScan project(Schema newSchema) {
+ throw new UnsupportedOperationException("Projection is not supported");
+ }
+
+ @Override
+ public CloseableIterable<PartitionStatistics> scan() {
+ if (snapshotId == null) {
+ if (table.currentSnapshot() == null) {
+ return CloseableIterable.of(List.of());
+ }
+
+ snapshotId = table.currentSnapshot().snapshotId();
+ }
+
+ Optional<PartitionStatisticsFile> statsFile =
+ table.partitionStatisticsFiles().stream()
+ .filter(f -> f.snapshotId() == snapshotId)
+ .findFirst();
+
+ if (statsFile.isEmpty()) {
+ return CloseableIterable.of(List.of());
+ }
+
+ Types.StructType partitionType = Partitioning.partitionType(table);
+ Schema schema = PartitionStatsHandler.schema(partitionType, TableUtil.formatVersion(table));
+
+ FileFormat fileFormat = FileFormat.fromFileName(statsFile.get().path());
Review Comment:
Move this discussion out of this 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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2541231898
##########
api/src/main/java/org/apache/iceberg/PartitionStatisticsScan.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+
+public interface PartitionStatisticsScan {
Review Comment:
nit: maybe at lease a oneliner javadoc
--
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
Re: [PR] PoC: API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#issuecomment-3744039435
This is split now into 3 parts ([PR](https://github.com/apache/iceberg/pull/14640) [PR](https://github.com/apache/iceberg/pull/14989) [PR](https://github.com/apache/iceberg/pull/14996)) and all of them are either merged or published, so this can be closed now.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2498262119
##########
api/src/main/java/org/apache/iceberg/PartitionStatisticsScan.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+
+public interface PartitionStatisticsScan {
+
+ /**
+ * Create a new scan from this scan's configuration that will use the given snapshot by ID.
+ *
+ * @param snapshotId a snapshot ID
+ * @return a new scan based on this with the given snapshot ID
+ * @throws IllegalArgumentException if the snapshot cannot be found
+ */
+ PartitionStatisticsScan useSnapshot(long snapshotId);
+
+ /**
+ * Create a new scan from the results of this, where partitions are filtered by the {@link
+ * Expression}.
+ *
+ * @param filter a filter expression
+ * @return a new scan based on this with results filtered by the expression
+ */
+ PartitionStatisticsScan filter(Expression filter);
+
+ /**
+ * Create a new scan from this with the schema as its projection.
+ *
+ * @param schema a projection schema
Review Comment:
How does the user create the `Schema`?
I would prefer something like the `DataFile` where the possible columns are available as constants, and the type is available as well. Maybe copy/move/deprecate the schema from the old place.
--
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
Re: [PR] PoC: API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab closed pull request #14508: PoC: API, Core, Spark: Scan API for partition stats
URL: https://github.com/apache/iceberg/pull/14508
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2511273797
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -225,7 +223,7 @@ public static PartitionStatisticsFile computeAndWriteStatsFile(Table table, long
}
try {
- stats = computeAndMergeStatsIncremental(table, snapshot, partitionType, statisticsFile);
+ stats = computeAndMergeStatsIncremental(table, snapshot, statisticsFile.snapshotId());
Review Comment:
> we produce this partitionType once for scanning the latest existing partition stats in case it's an incremental computation, and then we produce it one more time when we sort the stats.
This is why I asked. Previous code calculated this once and reused it later
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2498269535
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatisticsScan.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Optional;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatisticsScan implements PartitionStatisticsScan {
+
+ private final Table table;
+ private Long snapshotId;
+
+ public BasePartitionStatisticsScan(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ public PartitionStatisticsScan useSnapshot(long newSnapshotId) {
+ Preconditions.checkArgument(
+ table.snapshot(newSnapshotId) != null, "Cannot find snapshot with ID %s", newSnapshotId);
+
+ this.snapshotId = newSnapshotId;
+ return this;
+ }
+
+ @Override
+ public PartitionStatisticsScan filter(Expression newFilter) {
+ throw new UnsupportedOperationException("Filtering is not supported");
+ }
+
+ @Override
+ public PartitionStatisticsScan project(Schema newSchema) {
+ throw new UnsupportedOperationException("Projection is not supported");
+ }
+
+ @Override
+ public CloseableIterable<PartitionStatistics> scan() {
+ if (snapshotId == null) {
+ if (table.currentSnapshot() == null) {
+ return CloseableIterable.of(List.of());
+ }
+
+ snapshotId = table.currentSnapshot().snapshotId();
+ }
+
+ Optional<PartitionStatisticsFile> statsFile =
+ table.partitionStatisticsFiles().stream()
+ .filter(f -> f.snapshotId() == snapshotId)
+ .findFirst();
+
+ if (statsFile.isEmpty()) {
+ return CloseableIterable.of(List.of());
+ }
+
+ Types.StructType partitionType = Partitioning.partitionType(table);
+ Schema schema = PartitionStatsHandler.schema(partitionType, TableUtil.formatVersion(table));
+
+ FileFormat fileFormat = FileFormat.fromFileName(statsFile.get().path());
Review Comment:
I still think that getting the file format from the name is brittle. Maybe not in this PR, but I would love to have this fixed
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2498289276
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -287,7 +289,11 @@ public static CloseableIterable<PartitionStats> readPartitionStatsFile(
CloseableIterable<StructLike> records =
InternalData.read(fileFormat, inputFile).project(schema).build();
- return CloseableIterable.transform(records, PartitionStatsHandler::recordToPartitionStats);
+ return CloseableIterable.transform(
+ records,
+ rec -> {
+ return (PartitionStats) recordToPartitionStats(rec);
+ });
Review Comment:
Do we need the braces?
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2527814219
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatistics.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * 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;
+
+import org.apache.iceberg.avro.SupportsIndexProjection;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatistics extends SupportsIndexProjection
+ implements PartitionStatistics {
+
+ private StructLike partition;
+ private Integer specId;
+ private Long dataRecordCount;
+ private Integer dataFileCount;
+ private Long totalDataFileSizeInBytes;
+ private Long positionDeleteRecordCount;
+ private Integer positionDeleteFileCount;
+ private Long equalityDeleteRecordCount;
+ private Integer equalityDeleteFileCount;
+ private Long totalRecordCount; // null by default
+ private Long lastUpdatedAt; // null by default
+ private Long lastUpdatedSnapshotId; // null by default
+ private Integer dvCount;
+
+ private static final int STATS_COUNT = 13;
+
+ public BasePartitionStatistics(StructLike partition, int specId) {
+ super(STATS_COUNT);
+
+ this.partition = partition;
+ this.specId = specId;
+
+ dataRecordCount = 0L;
+ dataFileCount = 0;
+ totalDataFileSizeInBytes = 0L;
+ positionDeleteRecordCount = 0L;
+ positionDeleteFileCount = 0;
+ equalityDeleteRecordCount = 0L;
+ equalityDeleteFileCount = 0;
+ dvCount = 0;
+ }
+
+ /** Used by internal readers to instantiate this class with a projection schema. */
+ public BasePartitionStatistics(Types.StructType projection) {
+ super(STATS_COUNT);
+
+ // TODO: For implementing projection using a custom Schema here we have to call the other super
+ // constructor with the base Schema and the projection Schema.
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public Integer specId() {
+ return specId;
+ }
+
+ @Override
+ public Long dataRecordCount() {
+ return dataRecordCount;
+ }
+
+ @Override
+ public Integer dataFileCount() {
+ return dataFileCount;
+ }
+
+ @Override
+ public Long totalDataFileSizeInBytes() {
+ return totalDataFileSizeInBytes;
+ }
+
+ @Override
+ public Long positionDeleteRecordCount() {
+ return positionDeleteRecordCount;
+ }
+
+ @Override
+ public Integer positionDeleteFileCount() {
+ return positionDeleteFileCount;
+ }
+
+ @Override
+ public Long equalityDeleteRecordCount() {
+ return equalityDeleteRecordCount;
+ }
+
+ @Override
+ public Integer equalityDeleteFileCount() {
+ return equalityDeleteFileCount;
+ }
+
+ @Override
+ public Long totalRecords() {
+ return totalRecordCount;
+ }
+
+ @Override
+ public Long lastUpdatedAt() {
+ return lastUpdatedAt;
+ }
+
+ @Override
+ public Long lastUpdatedSnapshotId() {
+ return lastUpdatedSnapshotId;
+ }
+
+ @Override
+ public Integer dvCount() {
+ return dvCount;
+ }
+
+ /**
+ * Updates the partition stats from the data/delete file.
+ *
+ * @param file the {@link ContentFile} from the manifest entry.
+ * @param snapshot the snapshot corresponding to the live entry.
+ */
+ void liveEntry(ContentFile<?> file, Snapshot snapshot) {
+ Preconditions.checkArgument(file.specId() == specId, "Spec IDs must match");
+
+ switch (file.content()) {
+ case DATA:
+ this.dataRecordCount += file.recordCount();
+ this.dataFileCount += 1;
+ this.totalDataFileSizeInBytes += file.fileSizeInBytes();
+ break;
+ case POSITION_DELETES:
+ this.positionDeleteRecordCount += file.recordCount();
+ if (file.format() == FileFormat.PUFFIN) {
+ this.dvCount += 1;
+ } else {
+ this.positionDeleteFileCount += 1;
+ }
+
+ break;
+ case EQUALITY_DELETES:
+ this.equalityDeleteRecordCount += file.recordCount();
+ this.equalityDeleteFileCount += 1;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported file content type: " + file.content());
+ }
+
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+
+ // Note: Not computing the `TOTAL_RECORD_COUNT` for now as it needs scanning the data.
+ }
+
+ /**
+ * Updates the modified time and snapshot ID for the deleted manifest entry.
+ *
+ * @param snapshot the snapshot corresponding to the deleted manifest entry.
+ */
+ void deletedEntry(Snapshot snapshot) {
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+ }
+
+ /**
+ * Decrement the counters as it was included in the previous stats and updates the modified time
+ * and snapshot ID for the deleted manifest entry.
+ *
+ * @param snapshot the snapshot corresponding to the deleted manifest entry.
+ */
+ void deletedEntryForIncrementalCompute(ContentFile<?> file, Snapshot snapshot) {
+ Preconditions.checkArgument(file.specId() == specId, "Spec IDs must match");
+
+ switch (file.content()) {
+ case DATA:
+ this.dataRecordCount -= file.recordCount();
+ this.dataFileCount -= 1;
+ this.totalDataFileSizeInBytes -= file.fileSizeInBytes();
+ break;
+ case POSITION_DELETES:
+ this.positionDeleteRecordCount -= file.recordCount();
+ if (file.format() == FileFormat.PUFFIN) {
+ this.dvCount -= 1;
+ } else {
+ this.positionDeleteFileCount -= 1;
+ }
+
+ break;
+ case EQUALITY_DELETES:
+ this.equalityDeleteRecordCount -= file.recordCount();
+ this.equalityDeleteFileCount -= 1;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported file content type: " + file.content());
+ }
+
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+ }
+
+ /**
+ * Appends statistics from given entry to current entry.
+ *
+ * @param entry the entry from which statistics will be sourced.
+ */
+ void appendStats(PartitionStatistics entry) {
+ Preconditions.checkArgument(entry.specId() != null, "Invalid spec ID: mull");
+ Preconditions.checkArgument(entry.specId().equals(this.specId), "Spec IDs must match");
+
+ this.dataRecordCount += entry.dataRecordCount();
+ this.dataFileCount += entry.dataFileCount();
+ this.totalDataFileSizeInBytes += entry.totalDataFileSizeInBytes();
+ this.positionDeleteRecordCount += entry.positionDeleteRecordCount();
+ this.positionDeleteFileCount += entry.positionDeleteFileCount();
+ this.equalityDeleteRecordCount += entry.equalityDeleteRecordCount();
+ this.equalityDeleteFileCount += entry.equalityDeleteFileCount();
Review Comment:
They can't be null, because this is on the write path where we use the full V2 or V3 schema for read/write. Added a comment
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatistics.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * 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;
+
+import org.apache.iceberg.avro.SupportsIndexProjection;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatistics extends SupportsIndexProjection
+ implements PartitionStatistics {
+
+ private StructLike partition;
+ private Integer specId;
+ private Long dataRecordCount;
+ private Integer dataFileCount;
+ private Long totalDataFileSizeInBytes;
+ private Long positionDeleteRecordCount;
+ private Integer positionDeleteFileCount;
+ private Long equalityDeleteRecordCount;
+ private Integer equalityDeleteFileCount;
+ private Long totalRecordCount; // null by default
+ private Long lastUpdatedAt; // null by default
+ private Long lastUpdatedSnapshotId; // null by default
Review Comment:
You're right, these comments are misleading now. Originally, for the write path these were true, but now on the read path any of the stats can be null. Removed the comments.
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -275,7 +275,9 @@ static PartitionStatisticsFile writePartitionStatsFile(
*
* @param schema The {@link Schema} of the partition statistics file.
* @param inputFile An {@link InputFile} pointing to the partition stats file.
+ * @deprecated will be removed in 1.12.0, use {@link PartitionStatisticsScan} instead
*/
+ @Deprecated
public static CloseableIterable<PartitionStats> readPartitionStatsFile(
Review Comment:
I replaced all the calls with the new Scan API, including the tests. So no, now the tests exercise the new way. Since this is deprecated now, I found it's fine. LMK if I missed something.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2523152669
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -275,7 +275,9 @@ static PartitionStatisticsFile writePartitionStatsFile(
*
* @param schema The {@link Schema} of the partition statistics file.
* @param inputFile An {@link InputFile} pointing to the partition stats file.
+ * @deprecated will be removed in 1.12.0, use {@link PartitionStatisticsScan} instead
*/
+ @Deprecated
public static CloseableIterable<PartitionStats> readPartitionStatsFile(
Review Comment:
Do we still have tests which executing the old code path?
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2498305338
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatisticsScan.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Optional;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatisticsScan implements PartitionStatisticsScan {
+
+ private final Table table;
+ private Long snapshotId;
+
+ public BasePartitionStatisticsScan(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ public PartitionStatisticsScan useSnapshot(long newSnapshotId) {
+ Preconditions.checkArgument(
+ table.snapshot(newSnapshotId) != null, "Cannot find snapshot with ID %s", newSnapshotId);
+
+ this.snapshotId = newSnapshotId;
+ return this;
+ }
+
+ @Override
+ public PartitionStatisticsScan filter(Expression newFilter) {
+ throw new UnsupportedOperationException("Filtering is not supported");
+ }
+
+ @Override
+ public PartitionStatisticsScan project(Schema newSchema) {
+ throw new UnsupportedOperationException("Projection is not supported");
+ }
+
+ @Override
+ public CloseableIterable<PartitionStatistics> scan() {
Review Comment:
Do we have tests for this?
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2498285321
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -225,7 +223,7 @@ public static PartitionStatisticsFile computeAndWriteStatsFile(Table table, long
}
try {
- stats = computeAndMergeStatsIncremental(table, snapshot, partitionType, statisticsFile);
+ stats = computeAndMergeStatsIncremental(table, snapshot, statisticsFile.snapshotId());
Review Comment:
Why do we remove the `partitionType` parameter? We recalculate it later again for every file. Isn't that unnecessary?
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2541241772
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatistics.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * 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;
+
+import org.apache.iceberg.avro.SupportsIndexProjection;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatistics extends SupportsIndexProjection
+ implements PartitionStatistics {
+
+ private StructLike partition;
+ private Integer specId;
+ private Long dataRecordCount;
+ private Integer dataFileCount;
+ private Long totalDataFileSizeInBytes;
+ private Long positionDeleteRecordCount;
+ private Integer positionDeleteFileCount;
+ private Long equalityDeleteRecordCount;
+ private Integer equalityDeleteFileCount;
+ private Long totalRecordCount; // null by default
+ private Long lastUpdatedAt; // null by default
+ private Long lastUpdatedSnapshotId; // null by default
+ private Integer dvCount;
+
+ private static final int STATS_COUNT = 13;
+
+ public BasePartitionStatistics(StructLike partition, int specId) {
+ super(STATS_COUNT);
+
+ this.partition = partition;
+ this.specId = specId;
+
+ dataRecordCount = 0L;
+ dataFileCount = 0;
+ totalDataFileSizeInBytes = 0L;
+ positionDeleteRecordCount = 0L;
+ positionDeleteFileCount = 0;
+ equalityDeleteRecordCount = 0L;
+ equalityDeleteFileCount = 0;
+ dvCount = 0;
+ }
+
+ /** Used by internal readers to instantiate this class with a projection schema. */
+ public BasePartitionStatistics(Types.StructType projection) {
+ super(STATS_COUNT);
+
+ // TODO: For implementing projection using a custom Schema here we have to call the other super
+ // constructor with the base Schema and the projection Schema.
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public Integer specId() {
+ return specId;
+ }
+
+ @Override
+ public Long dataRecordCount() {
+ return dataRecordCount;
+ }
+
+ @Override
+ public Integer dataFileCount() {
+ return dataFileCount;
+ }
+
+ @Override
+ public Long totalDataFileSizeInBytes() {
+ return totalDataFileSizeInBytes;
+ }
+
+ @Override
+ public Long positionDeleteRecordCount() {
+ return positionDeleteRecordCount;
+ }
+
+ @Override
+ public Integer positionDeleteFileCount() {
+ return positionDeleteFileCount;
+ }
+
+ @Override
+ public Long equalityDeleteRecordCount() {
+ return equalityDeleteRecordCount;
+ }
+
+ @Override
+ public Integer equalityDeleteFileCount() {
+ return equalityDeleteFileCount;
+ }
+
+ @Override
+ public Long totalRecords() {
+ return totalRecordCount;
+ }
+
+ @Override
+ public Long lastUpdatedAt() {
+ return lastUpdatedAt;
+ }
+
+ @Override
+ public Long lastUpdatedSnapshotId() {
+ return lastUpdatedSnapshotId;
+ }
+
+ @Override
+ public Integer dvCount() {
+ return dvCount;
+ }
+
+ /**
+ * Updates the partition stats from the data/delete file.
+ *
+ * @param file the {@link ContentFile} from the manifest entry.
+ * @param snapshot the snapshot corresponding to the live entry.
+ */
+ void liveEntry(ContentFile<?> file, Snapshot snapshot) {
+ Preconditions.checkArgument(file.specId() == specId, "Spec IDs must match");
+
+ switch (file.content()) {
+ case DATA:
+ this.dataRecordCount += file.recordCount();
+ this.dataFileCount += 1;
+ this.totalDataFileSizeInBytes += file.fileSizeInBytes();
+ break;
+ case POSITION_DELETES:
+ this.positionDeleteRecordCount += file.recordCount();
+ if (file.format() == FileFormat.PUFFIN) {
+ this.dvCount += 1;
+ } else {
+ this.positionDeleteFileCount += 1;
+ }
+
+ break;
+ case EQUALITY_DELETES:
+ this.equalityDeleteRecordCount += file.recordCount();
+ this.equalityDeleteFileCount += 1;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported file content type: " + file.content());
+ }
+
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+
+ // Note: Not computing the `TOTAL_RECORD_COUNT` for now as it needs scanning the data.
+ }
+
+ /**
+ * Updates the modified time and snapshot ID for the deleted manifest entry.
+ *
+ * @param snapshot the snapshot corresponding to the deleted manifest entry.
+ */
+ void deletedEntry(Snapshot snapshot) {
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+ }
+
+ /**
+ * Decrement the counters as it was included in the previous stats and updates the modified time
+ * and snapshot ID for the deleted manifest entry.
+ *
+ * @param snapshot the snapshot corresponding to the deleted manifest entry.
+ */
+ void deletedEntryForIncrementalCompute(ContentFile<?> file, Snapshot snapshot) {
+ Preconditions.checkArgument(file.specId() == specId, "Spec IDs must match");
+
+ switch (file.content()) {
+ case DATA:
+ this.dataRecordCount -= file.recordCount();
+ this.dataFileCount -= 1;
+ this.totalDataFileSizeInBytes -= file.fileSizeInBytes();
+ break;
+ case POSITION_DELETES:
+ this.positionDeleteRecordCount -= file.recordCount();
+ if (file.format() == FileFormat.PUFFIN) {
+ this.dvCount -= 1;
+ } else {
+ this.positionDeleteFileCount -= 1;
+ }
+
+ break;
+ case EQUALITY_DELETES:
+ this.equalityDeleteRecordCount -= file.recordCount();
+ this.equalityDeleteFileCount -= 1;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported file content type: " + file.content());
+ }
+
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+ }
+
+ /**
+ * Appends statistics from given entry to current entry.
+ *
+ * @param entry the entry from which statistics will be sourced.
+ */
+ void appendStats(PartitionStatistics entry) {
+ Preconditions.checkArgument(entry.specId() != null, "Invalid spec ID: mull");
+ Preconditions.checkArgument(entry.specId().equals(this.specId), "Spec IDs must match");
+
+ this.dataRecordCount += entry.dataRecordCount();
+ this.dataFileCount += entry.dataFileCount();
+ this.totalDataFileSizeInBytes += entry.totalDataFileSizeInBytes();
+ this.positionDeleteRecordCount += entry.positionDeleteRecordCount();
+ this.positionDeleteFileCount += entry.positionDeleteFileCount();
+ this.equalityDeleteRecordCount += entry.equalityDeleteRecordCount();
+ this.equalityDeleteFileCount += entry.equalityDeleteFileCount();
Review Comment:
Maybe do a precondition check on them?
--
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
Re: [PR] PoC: API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#issuecomment-3718117118
Let's keep this open until all the relevant PRs are merged
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#issuecomment-3491094392
Some background: The current the way to query partition stats is through `PartitionStatsHandler.readPartitionStatsFile()`. For the the user has to put together the schema and get the input file to read. It would be beneficial for easier usability (also one comment on [my stats proposal doc](https://docs.google.com/document/d/1H9uYt53Q1_CcOXOfLcr0hXRxvqflg_k_xeVorMLrWbM/edit?pli=1&disco=AAABoDRdYcw) mentions) to have a more convenient API to scan partition stats. This could also have filter and projection capabilities.
The content of this PR:
1) Introduce `PartitionStatisticsScan` API and its implementation `BasePartitionStatisticsScan` in core. For simplicity this has the functionality that exists today, no filtering by partition, no projection.
2) Replace the usage of `PartitionStatsHandler.readPartitionStatsFile()` with the new API
3) Introduce PartitionStatistics interface into the API module, make PartitionStats in core to derive from this. This is needed so that the Scan API could use this as return value, while the existing PartitionStats class is in core module.
4) Replace the usage of PartitionStats whenever possible with the new interface.
These could possibly be some follow-up steps:
1) Implementation of filter() and project() on the new Scan API
2) The naming of affected classes is a bit weird: interface `api/PartitionStatistics` that is implemented by `core/PartitionStats`. Ideally the name of the implementation would be `BasePartitionStatistics`. As a next step we can introduce a class with the same content and new name and deprecate the existing one, also remove usage. Changes within PartitionStats are easier to review in case "renaming" happens in a follow-up PR.
3) Older Spark versions should be covered
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2494398563
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -349,7 +350,7 @@ private static Collection<PartitionStats> computeAndMergeStatsIncremental(
Pair.of(key.first(), partitionDataToRecord((PartitionData) key.second())),
value,
(existingEntry, newEntry) -> {
- existingEntry.appendStats(newEntry);
+ ((PartitionStats) existingEntry).appendStats(newEntry);
Review Comment:
If PartitionStatistics interface had the `appendStats` function, this cast (and another occurrence) wouldn't be needed. It seemed a bit weird to have it there, but I'm open to make this change to clean up casts.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2523130907
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatistics.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * 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;
+
+import org.apache.iceberg.avro.SupportsIndexProjection;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatistics extends SupportsIndexProjection
+ implements PartitionStatistics {
+
+ private StructLike partition;
+ private Integer specId;
+ private Long dataRecordCount;
+ private Integer dataFileCount;
+ private Long totalDataFileSizeInBytes;
+ private Long positionDeleteRecordCount;
+ private Integer positionDeleteFileCount;
+ private Long equalityDeleteRecordCount;
+ private Integer equalityDeleteFileCount;
+ private Long totalRecordCount; // null by default
+ private Long lastUpdatedAt; // null by default
+ private Long lastUpdatedSnapshotId; // null by default
+ private Integer dvCount;
+
+ private static final int STATS_COUNT = 13;
+
+ public BasePartitionStatistics(StructLike partition, int specId) {
+ super(STATS_COUNT);
+
+ this.partition = partition;
+ this.specId = specId;
+
+ dataRecordCount = 0L;
+ dataFileCount = 0;
+ totalDataFileSizeInBytes = 0L;
+ positionDeleteRecordCount = 0L;
+ positionDeleteFileCount = 0;
+ equalityDeleteRecordCount = 0L;
+ equalityDeleteFileCount = 0;
+ dvCount = 0;
+ }
+
+ /** Used by internal readers to instantiate this class with a projection schema. */
+ public BasePartitionStatistics(Types.StructType projection) {
+ super(STATS_COUNT);
+
+ // TODO: For implementing projection using a custom Schema here we have to call the other super
+ // constructor with the base Schema and the projection Schema.
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public Integer specId() {
+ return specId;
+ }
+
+ @Override
+ public Long dataRecordCount() {
+ return dataRecordCount;
+ }
+
+ @Override
+ public Integer dataFileCount() {
+ return dataFileCount;
+ }
+
+ @Override
+ public Long totalDataFileSizeInBytes() {
+ return totalDataFileSizeInBytes;
+ }
+
+ @Override
+ public Long positionDeleteRecordCount() {
+ return positionDeleteRecordCount;
+ }
+
+ @Override
+ public Integer positionDeleteFileCount() {
+ return positionDeleteFileCount;
+ }
+
+ @Override
+ public Long equalityDeleteRecordCount() {
+ return equalityDeleteRecordCount;
+ }
+
+ @Override
+ public Integer equalityDeleteFileCount() {
+ return equalityDeleteFileCount;
+ }
+
+ @Override
+ public Long totalRecords() {
+ return totalRecordCount;
+ }
+
+ @Override
+ public Long lastUpdatedAt() {
+ return lastUpdatedAt;
+ }
+
+ @Override
+ public Long lastUpdatedSnapshotId() {
+ return lastUpdatedSnapshotId;
+ }
+
+ @Override
+ public Integer dvCount() {
+ return dvCount;
+ }
+
+ /**
+ * Updates the partition stats from the data/delete file.
+ *
+ * @param file the {@link ContentFile} from the manifest entry.
+ * @param snapshot the snapshot corresponding to the live entry.
+ */
+ void liveEntry(ContentFile<?> file, Snapshot snapshot) {
+ Preconditions.checkArgument(file.specId() == specId, "Spec IDs must match");
+
+ switch (file.content()) {
+ case DATA:
+ this.dataRecordCount += file.recordCount();
+ this.dataFileCount += 1;
+ this.totalDataFileSizeInBytes += file.fileSizeInBytes();
+ break;
+ case POSITION_DELETES:
+ this.positionDeleteRecordCount += file.recordCount();
+ if (file.format() == FileFormat.PUFFIN) {
+ this.dvCount += 1;
+ } else {
+ this.positionDeleteFileCount += 1;
+ }
+
+ break;
+ case EQUALITY_DELETES:
+ this.equalityDeleteRecordCount += file.recordCount();
+ this.equalityDeleteFileCount += 1;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported file content type: " + file.content());
+ }
+
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+
+ // Note: Not computing the `TOTAL_RECORD_COUNT` for now as it needs scanning the data.
+ }
+
+ /**
+ * Updates the modified time and snapshot ID for the deleted manifest entry.
+ *
+ * @param snapshot the snapshot corresponding to the deleted manifest entry.
+ */
+ void deletedEntry(Snapshot snapshot) {
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+ }
+
+ /**
+ * Decrement the counters as it was included in the previous stats and updates the modified time
+ * and snapshot ID for the deleted manifest entry.
+ *
+ * @param snapshot the snapshot corresponding to the deleted manifest entry.
+ */
+ void deletedEntryForIncrementalCompute(ContentFile<?> file, Snapshot snapshot) {
+ Preconditions.checkArgument(file.specId() == specId, "Spec IDs must match");
+
+ switch (file.content()) {
+ case DATA:
+ this.dataRecordCount -= file.recordCount();
+ this.dataFileCount -= 1;
+ this.totalDataFileSizeInBytes -= file.fileSizeInBytes();
+ break;
+ case POSITION_DELETES:
+ this.positionDeleteRecordCount -= file.recordCount();
+ if (file.format() == FileFormat.PUFFIN) {
+ this.dvCount -= 1;
+ } else {
+ this.positionDeleteFileCount -= 1;
+ }
+
+ break;
+ case EQUALITY_DELETES:
+ this.equalityDeleteRecordCount -= file.recordCount();
+ this.equalityDeleteFileCount -= 1;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported file content type: " + file.content());
+ }
+
+ if (snapshot != null) {
+ updateSnapshotInfo(snapshot.snapshotId(), snapshot.timestampMillis());
+ }
+ }
+
+ /**
+ * Appends statistics from given entry to current entry.
+ *
+ * @param entry the entry from which statistics will be sourced.
+ */
+ void appendStats(PartitionStatistics entry) {
+ Preconditions.checkArgument(entry.specId() != null, "Invalid spec ID: mull");
+ Preconditions.checkArgument(entry.specId().equals(this.specId), "Spec IDs must match");
+
+ this.dataRecordCount += entry.dataRecordCount();
+ this.dataFileCount += entry.dataFileCount();
+ this.totalDataFileSizeInBytes += entry.totalDataFileSizeInBytes();
+ this.positionDeleteRecordCount += entry.positionDeleteRecordCount();
+ this.positionDeleteFileCount += entry.positionDeleteFileCount();
+ this.equalityDeleteRecordCount += entry.equalityDeleteRecordCount();
+ this.equalityDeleteFileCount += entry.equalityDeleteFileCount();
Review Comment:
What happens when one of these are null?
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2511131964
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -225,7 +223,7 @@ public static PartitionStatisticsFile computeAndWriteStatsFile(Table table, long
}
try {
- stats = computeAndMergeStatsIncremental(table, snapshot, partitionType, statisticsFile);
+ stats = computeAndMergeStatsIncremental(table, snapshot, statisticsFile.snapshotId());
Review Comment:
The reason this param is removed is that `computeAndMergeStatsIncremental` is changed to use the new Scan API instead of `readPartitionStatsFile`. Since the partitionType is deliberately made internal to the Scan, this is no longer needed in this function.
What Do you mean we recalculate it later for every file? Whenever we calculate stats, we produce this partitionType once for scanning the latest existing partition stats in case it's an incremental computation, and then we produce it one more time when we sort the stats. I think this is cheap enough to worth keeping it internal to the Scan API and not giving a way for users to inject it into the scan in order to save one computation of this.
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2510971424
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatisticsScan.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Optional;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatisticsScan implements PartitionStatisticsScan {
+
+ private final Table table;
+ private Long snapshotId;
+
+ public BasePartitionStatisticsScan(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ public PartitionStatisticsScan useSnapshot(long newSnapshotId) {
+ Preconditions.checkArgument(
+ table.snapshot(newSnapshotId) != null, "Cannot find snapshot with ID %s", newSnapshotId);
+
+ this.snapshotId = newSnapshotId;
+ return this;
+ }
+
+ @Override
+ public PartitionStatisticsScan filter(Expression newFilter) {
+ throw new UnsupportedOperationException("Filtering is not supported");
+ }
+
+ @Override
+ public PartitionStatisticsScan project(Schema newSchema) {
+ throw new UnsupportedOperationException("Projection is not supported");
+ }
+
+ @Override
+ public CloseableIterable<PartitionStatistics> scan() {
+ if (snapshotId == null) {
+ if (table.currentSnapshot() == null) {
+ return CloseableIterable.of(List.of());
+ }
+
+ snapshotId = table.currentSnapshot().snapshotId();
+ }
+
+ Optional<PartitionStatisticsFile> statsFile =
+ table.partitionStatisticsFiles().stream()
+ .filter(f -> f.snapshotId() == snapshotId)
+ .findFirst();
+
+ if (statsFile.isEmpty()) {
+ return CloseableIterable.of(List.of());
+ }
+
+ Types.StructType partitionType = Partitioning.partitionType(table);
+ Schema schema = PartitionStatsHandler.schema(partitionType, TableUtil.formatVersion(table));
+
+ FileFormat fileFormat = FileFormat.fromFileName(statsFile.get().path());
Review Comment:
I can take a look at this in a separate PR. Currently, `PartitionStatisticsFile` doesn't know the file format. We can extend that and force the internal writer to populate such field, however, since this is already released there might be tables where the additional info for file format isn't written, so we'd still need to keep the old logic there. Otherwise we'd break the stats for those tables.
Also, this would require a spec change too. Not sure, but would this mean that such an addition could only be part of V4?
--
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
Re: [PR] API, Core, Spark: Scan API for partition stats [iceberg]
Posted by "gaborkaszab (via GitHub)" <gi...@apache.org>.
gaborkaszab commented on code in PR #14508:
URL: https://github.com/apache/iceberg/pull/14508#discussion_r2513269762
##########
api/src/main/java/org/apache/iceberg/PartitionStatisticsScan.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+
+public interface PartitionStatisticsScan {
+
+ /**
+ * Create a new scan from this scan's configuration that will use the given snapshot by ID.
+ *
+ * @param snapshotId a snapshot ID
+ * @return a new scan based on this with the given snapshot ID
+ * @throws IllegalArgumentException if the snapshot cannot be found
+ */
+ PartitionStatisticsScan useSnapshot(long snapshotId);
+
+ /**
+ * Create a new scan from the results of this, where partitions are filtered by the {@link
+ * Expression}.
+ *
+ * @param filter a filter expression
+ * @return a new scan based on this with results filtered by the expression
+ */
+ PartitionStatisticsScan filter(Expression filter);
+
+ /**
+ * Create a new scan from this with the schema as its projection.
Review Comment:
Thank for pointing this out! My initial plan was to always query the 'traditional' partition stats and allow projection for the column-based once when we introduce them later. But it makes sense to project also the existing ones, and then the current design with PartitionStatistics isn't suitable for that. Let me wrap my head around this and come back with a different design that can tackle this too.
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -287,7 +289,11 @@ public static CloseableIterable<PartitionStats> readPartitionStatsFile(
CloseableIterable<StructLike> records =
InternalData.read(fileFormat, inputFile).project(schema).build();
- return CloseableIterable.transform(records, PartitionStatsHandler::recordToPartitionStats);
+ return CloseableIterable.transform(
+ records,
+ rec -> {
+ return (PartitionStats) recordToPartitionStats(rec);
+ });
Review Comment:
We don't. Removed
##########
core/src/main/java/org/apache/iceberg/BasePartitionStatisticsScan.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Optional;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+public class BasePartitionStatisticsScan implements PartitionStatisticsScan {
+
+ private final Table table;
+ private Long snapshotId;
+
+ public BasePartitionStatisticsScan(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ public PartitionStatisticsScan useSnapshot(long newSnapshotId) {
+ Preconditions.checkArgument(
+ table.snapshot(newSnapshotId) != null, "Cannot find snapshot with ID %s", newSnapshotId);
+
+ this.snapshotId = newSnapshotId;
+ return this;
+ }
+
+ @Override
+ public PartitionStatisticsScan filter(Expression newFilter) {
+ throw new UnsupportedOperationException("Filtering is not supported");
+ }
+
+ @Override
+ public PartitionStatisticsScan project(Schema newSchema) {
+ throw new UnsupportedOperationException("Projection is not supported");
+ }
+
+ @Override
+ public CloseableIterable<PartitionStatistics> scan() {
Review Comment:
I haven't introduced dedicated tests for this because now the capabilities is the same as we had before this patch, and TestPartitionStatsHandler covers it. Let's finalize the API part and I'll add a separate test suite too if you think it makes sense at this point.
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -225,7 +223,7 @@ public static PartitionStatisticsFile computeAndWriteStatsFile(Table table, long
}
try {
- stats = computeAndMergeStatsIncremental(table, snapshot, partitionType, statisticsFile);
+ stats = computeAndMergeStatsIncremental(table, snapshot, statisticsFile.snapshotId());
Review Comment:
The new Scan API encapsulates constructing a partitionType so no need to give it to the function as param. As a result later on we need the partitionType for sorting and writing, so we do calculate it. It seems cheap to produce so I don't see any issues with this.
--
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