You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2023/01/13 19:04:37 UTC
[GitHub] [iceberg] huaxingao opened a new pull request, #6582: Add a Spark procedure to collect NDV
huaxingao opened a new pull request, #6582:
URL: https://github.com/apache/iceberg/pull/6582
Add a Spark procedure to collect NDV, which will be used for CBO.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1072805278
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@findepi Thanks a lot for your comment. I am wondering why Theta sketch is preferred over HLL. Is it because Theta sketch works better for larger data, and has better set intersection and difference operation?
It would be ideal Spark can support Theta sketch, but it may take a long time for the support to get in. Before Theta sketch is available in Spark, if we can't put table level NDV in puffin file, is there a better place we can store 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
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1092522395
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
To be clear I don't think we need to get this into OSS Spark, I think it's fine if we generate these sketches in user land code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] flyrain commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
flyrain commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070140194
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
We should use one blob type for NDV ideally, although Spark doesn't have the sketch data. I'm also curious how sketch data is useful for a table level metric. It is absolutely useful for file-level and partition-level since we can merge them 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
[GitHub] [iceberg] flyrain commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1085728930
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
> Trino provides ANALYZE statement to analyze stats for a table, and it currently computes a Theta sketch and puts it in Iceberg table's Puffin stats file.
Is it possible to do that in Spark? @huaxingao. I'd be awesome if Spark can do that, otherwise we can fall back to approach 2.1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] findepi commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
findepi commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073253298
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
> I am wondering why Theta sketch is preferred over HLL. Is it because Theta sketch works better for larger data, and has better set intersection and difference operation?
Theta supports union, intersection and difference, HLL supports only union.
In my testing they were roughly similar in NDV quality for unions.
See results in https://github.com/trinodb/trino/pull/14290#issuecomment-1262160870
> I have a side question for puffin file: Is puffin file only used to store table level stats?
currently, yes
this is how Puffin is **currently** integrated in Iceberg spec.
there is ongoing work for having partition-level stats, see eg https://github.com/apache/iceberg/pull/1985 https://github.com/apache/iceberg/issues/1832 https://github.com/apache/iceberg/issues/1833
> I am currently taking a look at the file level bloom filter and wondering where I should save them
this comment from @rdblue looks relevant https://github.com/apache/iceberg/issues/1832#issuecomment-757072379
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] ajantha-bhat commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
ajantha-bhat commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073335902
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I think https://github.com/apache/iceberg/pull/1985 https://github.com/apache/iceberg/issues/1832 https://github.com/apache/iceberg/issues/1833 are super old. We are discussing partition stats in the below proposal now a days.
https://docs.google.com/document/d/1vaufuD47kMijz97LxM67X8OX-W2Wq7nmlz3jRo8J5Qk/edit?usp=sharing
**In this proposal, we are planning to store partition-level NDV stats also in a puffin file.** So, each partition will have one puffin file. These puffin file's location will become a cell in a row along with file count, row count in a general table-level statistics file (something like an index file) stored as sorted avro or parquet for every snapshot.
So, I am also interested to use the same CALL procedure to collect partition-level stats too (with a different argument).
Or maybe we need to step back and design some generic interfaces that can work for table level, partition level and even file level stats.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1069918267
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.procedures;
+
+import static java.lang.String.format;
+import static java.util.UUID.randomUUID;
+import static org.apache.hadoop.shaded.com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns
+ * and sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(format("%s.stats", randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
+ for (int i = 0; i < columnSizes; i++) {
+ writer.add(
+ new Blob(
+ StandardBlobTypes.NDV_BLOB,
+ ImmutableList.of(fieldId[i]),
+ table.currentSnapshot().snapshotId(),
+ table.currentSnapshot().sequenceNumber(),
+ ByteBuffer.allocate(0),
Review Comment:
Since Spark doesn't use Apache DataSketches to collect approximate NDV, I don't have a Sketch to set here. Set this to an empty `ByteBuffer` for 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
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1089308426
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@RussellSpitzer @findepi I agree it would be ideal if Spark can support Theta sketches. I will take a look to see the possibility to implement this.
@findepi Besides using NDV, Spark also uses other column stats such as NumOfNulls, Min, Max, etc. for CBO. I am wondering if Trino also use these stats and if these stats should also be stored in `TableMetaData`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] szehon-ho commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070150005
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(String.format("%s.stats", UUID.randomUUID()));
Review Comment:
If it exists, we throw FileNotFoundException? Should we just check and throw better exception?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(String.format("%s.stats", UUID.randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
+ for (int i = 0; i < columnSizes; i++) {
+ writer.add(
+ new Blob(
+ StandardBlobTypes.NDV_BLOB,
+ ImmutableList.of(fieldId[i]),
+ table.currentSnapshot().snapshotId(),
+ table.currentSnapshot().sequenceNumber(),
+ ByteBuffer.allocate(0),
+ null,
+ ImmutableMap.of("ndv", Long.toString(ndvs[i]))));
+ }
+ writer.finish();
+
+ GenericStatisticsFile statisticsFile =
+ new GenericStatisticsFile(
+ table.currentSnapshot().snapshotId(),
+ path,
+ writer.fileSize(),
+ writer.footerSize(),
+ writer.writtenBlobsMetadata().stream()
+ .map(GenericBlobMetadata::from)
+ .collect(ImmutableList.toImmutableList()));
+
+ table
+ .updateStatistics()
+ .setStatistics(table.currentSnapshot().snapshotId(), statisticsFile)
+ .commit();
+ } catch (IOException exception) {
+ throw new UncheckedIOException(exception);
+ }
+
+ String viewName = viewName(args, tableName);
+ // Create a view for users to query
+ df.createOrReplaceTempView(viewName);
Review Comment:
Not sure if I missed something, is there a point to keeping it as view, if its already returned by the procedure?
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
'blob' seems a bit redundant as they are all blobs? And also looking at the code, it's an approx ndv, which I didnt get from this name.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.procedures;
+
+import static java.lang.String.format;
+import static java.util.UUID.randomUUID;
+import static org.apache.hadoop.shaded.com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns
+ * and sets this to the table's StatisticsFile.
+ */
Review Comment:
Yea we have all those in the Iceberg file level metadata already, wonder if its necessary as we could combine those to have an aggregate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070181796
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@szehon-ho Right, we should have a better name for this. I am not sure if we can have a common blob type here. I will wait for @findepi 's input before changing this one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] flyrain commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
flyrain commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1081900515
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
The only downside is that we are going to have two items for one purpose.
1. A new blob type for NDV used by Spark
2. APACHE_DATASKETCHES_THETA_V1 for NDV used by Trino, and updated by Trino.
I believe that's something we are trying to avoid. I'm not sure why do we need a sketch blob for NDV at the table level. Spark only need a NDV number for its planning. Does Trino need something extra?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] ajantha-bhat commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1083604944
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
Review Comment:
@RussellSpitzer, @flyrain, @huaxingao: Is it good to have a spark action first and call that action from this procedure? This way the users who use only APIs can also leverage this feature.
--
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] Add a Spark procedure to collect NDV [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#issuecomment-2307946260
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
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1069901978
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
Spark doesn't use Apache DataSketches to collect approximate NDV, so I am adding a new blob type. Hope this is OK.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073864598
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I believe @findepi 's suggestion is to make fake Theta sketches (until we have real theta sketches)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073895130
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I am OK with fake Theta sketches. I will leave the blob with empty contents, and put NDV number in the property.
Should I still use `APACHE_DATASKETCHES_THETA_V1` as blob type?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] findepi commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
findepi commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1081236309
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
> I believe @findepi 's suggestion is to make fake Theta sketches (until we have real theta sketches)
Not exactly.
If you really want to fake something, it needs to be a different blob type.
> Should I still use APACHE_DATASKETCHES_THETA_V1 as blob type?
For real Theta sketches yes.
For anything "unreal", no. It would break incremental stat updates (Trino WIP https://github.com/trinodb/trino/pull/15441 for context)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] findepi commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
findepi commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1072412666
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
> Spark doesn't use Apache DataSketches to collect approximate NDV
Same was true for Trino. Trino uses HLL by default.
I introduced DataSketches Theta aggregation so that we can be compatible. For that I had to revamped stats collection SPI so that connectors can request desired sketch format, as previously it was hard-coded: if a connector wants NDV information they get a HLL sketch.
For NDV information, without update'able sketch, we shouldn't use a blob at all. The NDV number is just a property of actual updateabke sketch stored in the Puffin file. For a POC, you can use a fake blob with empty content, and associated NDV number as its property. Just give it some blob type name that reveals it's a fake temporary. For production use, we want to write Theta sketches.
cc @rdblue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] findepi commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1088814346
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I agree with you @RussellSpitzer . BTW engine interop is the primary reason why we have settled on Theta sketches. For Trino it would be easier to go with HLL, since that's what Trino engine & SPI are supporting for years 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
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070017872
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@findepi What are you using for NDV stats here? I figure we should have a common blob type
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070181906
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.procedures;
+
+import static java.lang.String.format;
+import static java.util.UUID.randomUUID;
+import static org.apache.hadoop.shaded.com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns
+ * and sets this to the table's StatisticsFile.
+ */
Review Comment:
We have file level metadata for max, min, num_nulls etc. That's why I was hesitate to include those here. We don't have file level ndv, though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] findepi commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1085085232
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
> Does Trino update the NDV sketch every time a write happens?
Not yet, but there is a WIP PR for that: https://github.com/trinodb/trino/pull/15441
> What if a table is wrote both by Trino and Spark? I believe the update from Spark side will be missing in that case.
That's unfortunately true, but we hope this is just a temporary limitation.
I would feel uncomfortably advising users not to use Spark because it cannot update Iceberg stats properly.
> A-synchronized operation like this procedure.
You need this anyway, since not all writes will update stats. For example, it's quite hard to updates NDV stats for a deletion (was this the _only_ appearance of a value, or one of many?)
Trino provides ANALYZE statement to analyze stats for a table, and it currently computes a Theta sketch and puts it in Iceberg table's Puffin stats file.
Automatic stats update will work well for append-only tables that do not undergo deletions or updates (or have deletions and updates in ignorable quantity).
--
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] Add a Spark procedure to collect NDV [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1619092149
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.procedures;
+
+import static java.lang.String.format;
+import static java.util.UUID.randomUUID;
+import static org.apache.hadoop.shaded.com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns
+ * and sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(format("%s.stats", randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
+ for (int i = 0; i < columnSizes; i++) {
+ writer.add(
+ new Blob(
+ StandardBlobTypes.NDV_BLOB,
+ ImmutableList.of(fieldId[i]),
+ table.currentSnapshot().snapshotId(),
+ table.currentSnapshot().sequenceNumber(),
+ ByteBuffer.allocate(0),
Review Comment:
I think Spark only adds support for HLL sketch, but we want to use Theta sketch.
Karuppayya is continuing the work. Here is the new PR https://github.com/apache/iceberg/pull/10288
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070182092
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(String.format("%s.stats", UUID.randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
+ for (int i = 0; i < columnSizes; i++) {
+ writer.add(
+ new Blob(
+ StandardBlobTypes.NDV_BLOB,
+ ImmutableList.of(fieldId[i]),
+ table.currentSnapshot().snapshotId(),
+ table.currentSnapshot().sequenceNumber(),
+ ByteBuffer.allocate(0),
+ null,
+ ImmutableMap.of("ndv", Long.toString(ndvs[i]))));
+ }
+ writer.finish();
+
+ GenericStatisticsFile statisticsFile =
+ new GenericStatisticsFile(
+ table.currentSnapshot().snapshotId(),
+ path,
+ writer.fileSize(),
+ writer.footerSize(),
+ writer.writtenBlobsMetadata().stream()
+ .map(GenericBlobMetadata::from)
+ .collect(ImmutableList.toImmutableList()));
+
+ table
+ .updateStatistics()
+ .setStatistics(table.currentSnapshot().snapshotId(), statisticsFile)
+ .commit();
+ } catch (IOException exception) {
+ throw new UncheckedIOException(exception);
+ }
+
+ String viewName = viewName(args, tableName);
+ // Create a view for users to query
+ df.createOrReplaceTempView(viewName);
Review Comment:
I kept this as a view so users will have an easy way to query the statistics information after calling the stored procedure.
The main reason I am adding this store procedure is because I can't get an agreement to implement `ANALYZE TABLE` for data source V2 in Spark. This stored procedure is doing something similar to `ANALYZE TABLE`. Normally after users analyze table, they will `DESCRIBE` to get the statistics information. I create a view so users can query the statistics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1081568857
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
OK... Then is everybody OK with adding a new blob type for ndv (without the real theta sketch)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073861447
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@rdblue Spark only needs NDV values from Iceberg.
The problem is that I don't have a place to save the table level NDV values. I initially thought I could save the NDV values as blobs in puffin file, but based on the discussion, it doesn't seem to be the correct way.
So I am still looking for the right place to save the table level NDV values...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070181959
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(String.format("%s.stats", UUID.randomUUID()));
Review Comment:
You mean `AlreadyExistsException`, right? Yes, we should check. I guess we can probably keep the `AlreadyExistsException` but make the error message better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1069914737
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.procedures;
+
+import static java.lang.String.format;
+import static java.util.UUID.randomUUID;
+import static org.apache.hadoop.shaded.com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns
+ * and sets this to the table's StatisticsFile.
+ */
Review Comment:
I am debating myself if I should collect ndv only or also collect everything else such as max, min, num_nulls etc. in ANALYZE TABLE. I will just collect ndv for 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
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070023914
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(String.format("%s.stats", UUID.randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
+ for (int i = 0; i < columnSizes; i++) {
+ writer.add(
+ new Blob(
+ StandardBlobTypes.NDV_BLOB,
Review Comment:
The issue with defining a new a new blob type here is we probably need to describe in the spec, otherwise folks won't be able to deserialize it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073878440
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@RussellSpitzer, it depends on whether Spark actually needs the sketch data. If it for some reason uses the HLL buffers, then we will want to track them and pass them in. I suspect that it doesn't need the buffers, in which case we would probably want to use Theta sketches for compatibility and incremental update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1080907485
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I don't think Spark supports incremental approx count distinct yet.
--
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] Add a Spark procedure to collect NDV [iceberg]
Posted by "rice668 (via GitHub)" <gi...@apache.org>.
rice668 commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1618744991
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.procedures;
+
+import static java.lang.String.format;
+import static java.util.UUID.randomUUID;
+import static org.apache.hadoop.shaded.com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns
+ * and sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(format("%s.stats", randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
+ for (int i = 0; i < columnSizes; i++) {
+ writer.add(
+ new Blob(
+ StandardBlobTypes.NDV_BLOB,
+ ImmutableList.of(fieldId[i]),
+ table.currentSnapshot().snapshotId(),
+ table.currentSnapshot().sequenceNumber(),
+ ByteBuffer.allocate(0),
Review Comment:
Hi, @huaxingao Spark in 3.5 now use Apache DataSketches to collect approximate NDV. See : https://www.databricks.com/blog/apache-spark-3-apache-datasketches-new-sketch-based-approximate-distinct-counting, and by the way, what is the status 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
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1072976064
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@findepi @rdblue
I have a side question for puffin file: Is puffin file only used to store table level stats?
I am currently taking a look at the file level bloom filter and wondering where I should save them. Right now we only have row group bloom filter support. If we can add file level bloom filter support, then at planning time, if a file doesn't contain the value we are looking for, we can use bloom filter to filter out this file and don't include it in `FileScanTask`. I was originally thinking of adding the file level bloom filters as blobs in puffin file. However, from the discussion we are having on NDV, I feel that puffin file is only intended for table level stats. If this is true, do we have a place to put file level stats that is large (e.g. bloom filter)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
RussellSpitzer commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070020289
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
Review Comment:
Since we are technically not using distinct here, maybe we should be calling the procedure "analyze"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
rdblue commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073829241
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
Focusing on just the table level NDV case:
* We don't want to introduce a new stats blob type if possible. We chose Theta because of the additional capabilities the sketch has.
* Can you be more specific about what Spark supports? Is it that we want HLL to feed HLL buffers into Spark? I would expect Spark to want NDV values, not the buffer/blob data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1084457847
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
Thanks @flyrain @findepi
I feel that we can only take approach 2.1 for now.
Spark doesn't support incrementally updatable NDV yet. I don't think this will be supported in the near future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1085864259
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I agree that it would be idea if Spark can support incrementally updatable NDV using Theta sketch, but i don't think this will happen soon.
I will take approach 2.1 for now if no one opposes 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
[GitHub] [iceberg] findepi commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1083735566
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@flyrain Trino needs NDV number for planning. the sketch is so that we can incrementally update the stats when writing new data to the table.
--
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] Add a Spark procedure to collect NDV [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#issuecomment-2345009488
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add a Spark procedure to collect NDV [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #6582: Add a Spark procedure to collect NDV
URL: https://github.com/apache/iceberg/pull/6582
--
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] Add a Spark procedure to collect NDV [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#issuecomment-2112165738
I saw a new PR on the same : https://github.com/apache/iceberg/pull/10288
--
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] Add a Spark procedure to collect NDV [iceberg]
Posted by "rice668 (via GitHub)" <gi...@apache.org>.
rice668 commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1618744991
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.procedures;
+
+import static java.lang.String.format;
+import static java.util.UUID.randomUUID;
+import static org.apache.hadoop.shaded.com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns
+ * and sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(format("%s.stats", randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
+ for (int i = 0; i < columnSizes; i++) {
+ writer.add(
+ new Blob(
+ StandardBlobTypes.NDV_BLOB,
+ ImmutableList.of(fieldId[i]),
+ table.currentSnapshot().snapshotId(),
+ table.currentSnapshot().sequenceNumber(),
+ ByteBuffer.allocate(0),
Review Comment:
Hi, @huaxingao Spark in 3.5 now use Apache DataSketches to collect approximate NDV. See : https://www.databricks.com/blog/apache-spark-3-apache-datasketches-new-sketch-based-approximate-distinct-counting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] ajantha-bhat commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
ajantha-bhat commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073335902
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I think https://github.com/apache/iceberg/pull/1985 https://github.com/apache/iceberg/issues/1832 https://github.com/apache/iceberg/issues/1833 are super old. We are discussing partition stats in the below proposal now a days.
https://docs.google.com/document/d/1vaufuD47kMijz97LxM67X8OX-W2Wq7nmlz3jRo8J5Qk/edit?usp=sharing
**In this proposal, we are planning to store partition-level NDV stats also in a puffin file.** So, each partition will have one puffin file. These puffin file's location will become a cell in a row along with file count, row count in a general statistics file (something like an index file) stored as sorted avro or parquet.
So, I am also interested to use the same CALL procedure to collect partition-level stats too (with a different argument).
Or maybe we need to step back and design some generic interfaces that can work for table level, partition level and even file level stats.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] ajantha-bhat commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
ajantha-bhat commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1080799876
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
@huaxingao: I am just wondering why are we not using the real sketches and update them incrementally. Isn't calling `APPROX_COUNT_DISTINCT` to compute freshly every time gonna be expensive in the long run?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1073895130
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I am OK with fake Theta sketches. I will leave the blob with empty contents, and put NDV number in the property.
Should I will still use `APACHE_DATASKETCHES_THETA_V1` as blob type?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] flyrain commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1084397164
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
Review Comment:
That sounds reasonable. `RewriteManifestsProcedure` did the same thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] flyrain commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1084414607
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
Thanks @findepi. Does Trino update the NDV sketch every time a write happens? What if a table is wrote both by Trino and Spark? I believe the update from Spark side will be missing in that case.
There are two general directions.
1. Update the metrics in each write commit. This keeps NDV alway updated, and no extra async operation is needed. It requires changes of each writer across engines. It's not a trivial effort.
2. A-synchronized operation like this procedure.
2.1. Not use the same type, only output the NDV number.
2.2 Use the same type, update the sketch as well.
If we'd pick the option 2. I feel like 2.2 could be a solution, so that, the metric is shared by both Spark and Trino, and both engines can update it. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1092379542
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
I think we should create real Theta sketches. If Spark only needs the NDV integer, then that's great. We can either keep track of NDV sketch and incrementally update internal to Iceberg, or we can do it async. Either way, there should be no need for a different sketch type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] huaxingao commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by GitBox <gi...@apache.org>.
huaxingao commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1070099281
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
Review Comment:
I will change the procedure name from `DistinctCountProcedure` to `AnalyzeTableProcedure`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1088030207
##########
core/src/main/java/org/apache/iceberg/puffin/StandardBlobTypes.java:
##########
@@ -26,4 +26,6 @@ private StandardBlobTypes() {}
* href="https://datasketches.apache.org/">Apache DataSketches</a> library
*/
public static final String APACHE_DATASKETCHES_THETA_V1 = "apache-datasketches-theta-v1";
+
+ public static final String NDV_BLOB = "ndv-blob";
Review Comment:
So it is impossible for us to make Theta Sketches using Spark? Things like that would be healthier for the long run if we implement that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] amogh-jahagirdar commented on a diff in pull request #6582: Add a Spark procedure to collect NDV
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #6582:
URL: https://github.com/apache/iceberg/pull/6582#discussion_r1082822392
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/DistinctCountProcedure.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.procedures;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import org.apache.iceberg.GenericBlobMetadata;
+import org.apache.iceberg.GenericStatisticsFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A procedure that gets approximate NDV (number of distinct value) for the requested columns and
+ * sets this to the table's StatisticsFile.
+ */
+public class DistinctCountProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(DistinctCountProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", DataTypes.StringType),
+ ProcedureParameter.optional("distinct_count_view", DataTypes.StringType),
+ ProcedureParameter.optional("columns", STRING_ARRAY),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("view_name", DataTypes.StringType, false, Metadata.empty())
+ });
+
+ public static SparkProcedures.ProcedureBuilder builder() {
+ return new Builder<DistinctCountProcedure>() {
+ @Override
+ protected DistinctCountProcedure doBuild() {
+ return new DistinctCountProcedure(tableCatalog());
+ }
+ };
+ }
+
+ private DistinctCountProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String tableName = args.getString(0);
+ Identifier tableIdent = toIdentifier(tableName, PARAMETERS[0].name());
+ SparkTable sparkTable = loadSparkTable(tableIdent);
+ StructType schema = sparkTable.schema();
+ Table table = sparkTable.table();
+ ArrayData columns = args.getArray(2);
+ int columnSizes = columns.numElements();
+
+ long[] ndvs = new long[columnSizes];
+ int[] fieldId = new int[columnSizes];
+ String query = "SELECT ";
+ for (int i = 0; i < columnSizes; i++) {
+ String colName = columns.getUTF8String(i).toString();
+ query += "APPROX_COUNT_DISTINCT(" + colName + "), ";
+ fieldId[i] = schema.fieldIndex(colName);
+ }
+
+ query = query.substring(0, query.length() - 2) + " FROM " + tableName;
+ Dataset<Row> df = spark().sql(query);
+
+ for (int i = 0; i < columnSizes; i++) {
+ ndvs[i] = df.head().getLong(i);
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ FileIO fileIO = ((HasTableOperations) table).operations().io();
+ String path = operations.metadataFileLocation(String.format("%s.stats", UUID.randomUUID()));
+ OutputFile outputFile = fileIO.newOutputFile(path);
+
+ try (PuffinWriter writer =
+ Puffin.write(outputFile).createdBy("Spark DistinctCountProcedure").build()) {
Review Comment:
Nit: Can we move "Spark DistinctCountProcedure" to a separate constant?
--
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