This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 19a6ef2f3 feat(table): add Theta sketch NDV helper (#2347)
19a6ef2f3 is described below
commit 19a6ef2f377fe88be2b9b41d06561b331cf2f290
Author: Tanmay Rauth <[email protected]>
AuthorDate: Thu Nov 5 09:11:34 2026 -0800
feat(table): add Theta sketch NDV helper (#2347)
* feat(table): add Theta sketch NDV helper
Add table/stats with a DataSketches Theta-sketch wrapper for computing
number-of-distinct-values, mutually readable with the sketches
Iceberg-Java stores in apache-datasketches-theta-v1 Puffin blobs. Pins
the same seed (9001), lgK (12), and compact serialization, skips empty
values as Java does, and produces results through a K-trimmed union.
Parity is covered by fixtures generated from Iceberg-Java's
ThetaSketchAgg path (dev/theta-fixtures), compared on decoded sketch
state per type.
* docs(table): note Sketch result methods rebuild a union per call
---
dev/theta-fixtures/.gitignore | 18 +
dev/theta-fixtures/GenerateThetaFixtures.java | 201 +++++++++
dev/theta-fixtures/README.md | 57 +++
dev/theta-fixtures/generate.sh | 87 ++++
go.mod | 1 +
go.sum | 2 +
table/stats/sketch.go | 241 +++++++++++
table/stats/sketch_test.go | 456 +++++++++++++++++++++
table/stats/testdata/README.md | 56 +++
table/stats/testdata/binary.sk | Bin 0 -> 2408 bytes
table/stats/testdata/boolean.sk | Bin 0 -> 32 bytes
table/stats/testdata/date.sk | Bin 0 -> 3216 bytes
table/stats/testdata/decimal.sk | Bin 0 -> 4672 bytes
table/stats/testdata/double.sk | Bin 0 -> 4672 bytes
table/stats/testdata/float.sk | Bin 0 -> 4672 bytes
table/stats/testdata/int.sk | Bin 0 -> 4672 bytes
table/stats/testdata/long.sk | Bin 0 -> 4672 bytes
table/stats/testdata/merge_exact_partitions.sk | Bin 0 -> 32792 bytes
table/stats/testdata/multi_partition_estimation.sk | Bin 0 -> 32792 bytes
table/stats/testdata/null_only.sk | Bin 0 -> 8 bytes
table/stats/testdata/single_partition_at_k.sk | Bin 0 -> 32784 bytes
.../stats/testdata/single_partition_estimation.sk | Bin 0 -> 32752 bytes
table/stats/testdata/string.sk | Bin 0 -> 4008 bytes
table/stats/testdata/string_empty_only.sk | Bin 0 -> 8 bytes
table/stats/testdata/timestamp.sk | Bin 0 -> 6640 bytes
table/stats/testdata/uuid.sk | Bin 0 -> 4672 bytes
26 files changed, 1119 insertions(+)
diff --git a/dev/theta-fixtures/.gitignore b/dev/theta-fixtures/.gitignore
new file mode 100644
index 000000000..30233e4bc
--- /dev/null
+++ b/dev/theta-fixtures/.gitignore
@@ -0,0 +1,18 @@
+# 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.
+
+build/
\ No newline at end of file
diff --git a/dev/theta-fixtures/GenerateThetaFixtures.java
b/dev/theta-fixtures/GenerateThetaFixtures.java
new file mode 100644
index 000000000..ccfed4799
--- /dev/null
+++ b/dev/theta-fixtures/GenerateThetaFixtures.java
@@ -0,0 +1,201 @@
+/*
+ * 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.
+ */
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.function.IntFunction;
+import org.apache.datasketches.common.Family;
+import org.apache.datasketches.memory.Memory;
+import org.apache.datasketches.theta.CompactSketch;
+import org.apache.datasketches.theta.SetOperationBuilder;
+import org.apache.datasketches.theta.Sketch;
+import org.apache.datasketches.theta.UpdateSketch;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+/**
+ * Produces Theta sketch fixtures the way Iceberg-Java's Spark
ComputeTableStats action does,
+ * for iceberg-go's parity tests in {@code table/stats/sketch_test.go}.
+ *
+ * <p>Each fixture replays the Spark aggregation in {@code
+ * org.apache.spark.sql.stats.ThetaSketchAgg} without a Spark runtime:
+ *
+ * <ul>
+ * <li>one {@code UpdateSketch.builder.setFamily(Family.ALPHA).build()}
buffer per partition;
+ * <li>non-null values fed through {@code
update(Conversions.toByteBuffer(type, value))};
+ * <li>each partial serialized with {@code compact().toByteArray()} and read
back with {@code
+ * CompactSketch.wrap};
+ * <li>partials merged into a fresh Alpha buffer with {@code new
+ * SetOperationBuilder().buildUnion.union(buffer, input)};
+ * <li>the result written with {@code compact().toByteArray()}, as {@code
NDVSketchUtil} stores
+ * it in the Puffin blob (before Puffin-level compression).
+ * </ul>
+ *
+ * <p>Every value generator here is mirrored in the Go test; keep the two in
sync.
+ */
+public class GenerateThetaFixtures {
+
+ private static final long MICROS_BASE = 1_600_000_000_000_000L;
+
+ public static void main(String[] args) throws IOException {
+ Path dir = Paths.get(System.getProperty("theta.fixtures.outputDir",
"build/theta-fixtures"));
+ Files.createDirectories(dir);
+
+ // All nulls: the sketch stays empty.
+ write(dir, "null_only", Types.LongType.get(), 1, 3, i -> null);
+
+ // One fixture per primitive type, hashed from Conversions.toByteBuffer.
Rows are dealt
+ // round-robin across partitions so the union merge path runs, with
duplicates and nulls.
+ write(dir, "boolean", Types.BooleanType.get(), 3, 100, i -> nullEvery(i,
7, i % 2 == 0));
+ write(dir, "int", Types.IntegerType.get(), 3, 1000, i -> nullEvery(i, 11,
(i % 600) * 7 - 2000));
+ write(dir, "long", Types.LongType.get(), 3, 1000,
+ i -> nullEvery(i, 11, (long) (i % 600) * 1_000_003L - 5_000_000_000L));
+ write(dir, "float", Types.FloatType.get(), 3, 1000, i -> nullEvery(i, 11,
floatValue(i)));
+ write(dir, "double", Types.DoubleType.get(), 3, 1000, i -> nullEvery(i,
11, (i % 600) / 3.0));
+ write(dir, "date", Types.DateType.get(), 3, 1000, i -> nullEvery(i, 11,
18000 + i % 400));
+ write(dir, "timestamp", Types.TimestampType.withoutZone(), 3, 1000,
+ i -> nullEvery(i, 11, MICROS_BASE + (long) (i % 900) * 1_000_000L));
+ write(dir, "string", Types.StringType.get(), 3, 1000, i -> nullEvery(i,
11, stringValue(i)));
+ write(dir, "binary", Types.BinaryType.get(), 3, 1000, i -> nullEvery(i,
11, binaryValue(i)));
+ write(dir, "decimal", Types.DecimalType.of(20, 2), 3, 1000,
+ i -> nullEvery(i, 11, new BigDecimal(BigInteger.valueOf((long) (i %
600 - 300) * 12345L), 2)));
+ write(dir, "uuid", Types.UUIDType.get(), 3, 1000,
+ i -> nullEvery(i, 11, new UUID(i % 600, (long) (i % 600) * 31L + 7L)));
+
+ // Only empty values: Java skips them, so the sketch stays empty.
+ write(dir, "string_empty_only", Types.StringType.get(), 2, 10, i -> "");
+
+ // Every partition stays in exact mode, but the union holds more than 4096
distinct values
+ // and trims to k.
+ writePartitioned(dir, "merge_exact_partitions", 4, p -> range(p * 1200L,
1500));
+
+ // One partition at exactly k distinct values, the largest that stays
exact.
+ writePartitioned(dir, "single_partition_at_k", 1, p -> range(0, 4096));
+
+ // One partition past k: the Alpha sketch enters estimation mode before
the final union.
+ writePartitioned(dir, "single_partition_estimation", 1, p -> range(0,
6000));
+
+ // Several estimation-mode partitions with overlapping ranges.
+ writePartitioned(dir, "multi_partition_estimation", 3, p -> range(p *
3000L, 6000));
+ }
+
+ private static Object nullEvery(int i, int every, Object value) {
+ return i % every == every - 1 ? null : value;
+ }
+
+ private static Object floatValue(int i) {
+ switch (i % 600) {
+ case 0:
+ return 0.0f;
+ case 1:
+ return -0.0f;
+ default:
+ return (i % 600) * 0.5f + 1f;
+ }
+ }
+
+ private static Object stringValue(int i) {
+ switch (i % 500) {
+ case 0:
+ return "";
+ case 1:
+ return "é";
+ default:
+ return "value-" + (i % 500);
+ }
+ }
+
+ private static Object binaryValue(int i) {
+ if (i % 300 == 0) {
+ return ByteBuffer.allocate(0);
+ }
+ return ByteBuffer.wrap(new byte[] {(byte) (i % 300), (byte) (i % 300 /
7)});
+ }
+
+ private static List<Object> range(long start, int count) {
+ List<Object> values = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ values.add(start + i);
+ }
+ return values;
+ }
+
+ /** Deals rows 0..numRows-1 round-robin across numPartitions partitions. */
+ private static void write(
+ Path dir, String name, Type type, int numPartitions, int numRows,
IntFunction<Object> value)
+ throws IOException {
+ writePartitioned(
+ dir,
+ name,
+ type,
+ numPartitions,
+ p -> {
+ List<Object> values = new ArrayList<>();
+ for (int i = p; i < numRows; i += numPartitions) {
+ values.add(value.apply(i));
+ }
+ return values;
+ });
+ }
+
+ private static void writePartitioned(
+ Path dir, String name, int numPartitions, IntFunction<List<Object>>
partition)
+ throws IOException {
+ writePartitioned(dir, name, Types.LongType.get(), numPartitions,
partition);
+ }
+
+ private static void writePartitioned(
+ Path dir, String name, Type type, int numPartitions,
IntFunction<List<Object>> partition)
+ throws IOException {
+ // ThetaSketchAgg.createAggregationBuffer / update / serialize, once per
partition.
+ List<Sketch> partials = new ArrayList<>();
+ for (int p = 0; p < numPartitions; p++) {
+ UpdateSketch buffer =
UpdateSketch.builder().setFamily(Family.ALPHA).build();
+ for (Object value : partition.apply(p)) {
+ if (value != null) {
+ buffer.update(Conversions.toByteBuffer(type, value));
+ }
+ }
+
partials.add(CompactSketch.wrap(Memory.wrap(buffer.compact().toByteArray())));
+ }
+
+ // ThetaSketchAgg final aggregation: a fresh buffer merged with every
partial.
+ Sketch result = UpdateSketch.builder().setFamily(Family.ALPHA).build();
+ for (Sketch partial : partials) {
+ result = new SetOperationBuilder().buildUnion().union(result, partial);
+ }
+
+ // NDVSketchUtil: the blob holds the wrapped final sketch's toByteArray().
+ Sketch blob =
CompactSketch.wrap(Memory.wrap(result.compact().toByteArray()));
+ byte[] bytes = blob.toByteArray();
+ Files.write(dir.resolve(name + ".sk"), bytes);
+ System.out.printf(
+ "%-28s ndv=%-6d retained=%-5d theta=%d bytes=%d%n",
+ name, (long) blob.getEstimate(), blob.getRetainedEntries(),
blob.getThetaLong(), bytes.length);
+ }
+}
diff --git a/dev/theta-fixtures/README.md b/dev/theta-fixtures/README.md
new file mode 100644
index 000000000..22d431feb
--- /dev/null
+++ b/dev/theta-fixtures/README.md
@@ -0,0 +1,57 @@
+<!--
+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.
+-->
+
+# Theta sketch fixture generator
+
+`GenerateThetaFixtures.java` produces the Theta sketch fixtures in
+`table/stats/testdata/` used by iceberg-go's NDV parity tests in
+`table/stats/sketch_test.go`.
+
+It replays the aggregation in Iceberg-Java's
+`org.apache.spark.sql.stats.ThetaSketchAgg` without a Spark runtime: an
+`UpdateSketch.builder.setFamily(Family.ALPHA)` sketch per partition, non-null
+values fed through `update(Conversions.toByteBuffer(type, value))`, partials
+serialized with `compact().toByteArray()` and merged into a fresh buffer with
+`SetOperationBuilder().buildUnion().union(buffer, input)`, and the result
stored
+as `NDVSketchUtil` stores it in the Puffin blob (before Puffin compression).
+
+The value generators are mirrored in `table/stats/sketch_test.go`; change both
+together.
+
+## Regenerate
+
+1. Build the Iceberg jars once in your `apache/iceberg` checkout and make sure
+ Gradle has resolved the Spark module's dependencies, which include
+ `datasketches-java`:
+
+ ```bash
+ ./gradlew :iceberg-api:jar :iceberg-bundled-guava:jar
+ ```
+
+2. From the iceberg-go repo root:
+
+ ```bash
+ ICEBERG_REPO=/path/to/iceberg dev/theta-fixtures/generate.sh
+ ```
+
+The script reads the DataSketches version from Iceberg's
+`gradle/libs.versions.toml`, compiles and runs the generator with `javac` and
+`java` (no Gradle), and copies the `.sk` files into `table/stats/testdata/`.
+The output is deterministic, so regenerating without changes leaves the
+fixtures byte-identical.
diff --git a/dev/theta-fixtures/generate.sh b/dev/theta-fixtures/generate.sh
new file mode 100755
index 000000000..a1ea1ed83
--- /dev/null
+++ b/dev/theta-fixtures/generate.sh
@@ -0,0 +1,87 @@
+#!/usr/bin/env bash
+# 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.
+
+# Build and run GenerateThetaFixtures.java directly via javac + java (no
Gradle).
+# Requires that apache/iceberg has been built once via `./gradlew
:iceberg-api:jar`
+# and that Gradle has downloaded the datasketches dependency Iceberg pins.
+
+set -euo pipefail
+
+ICEBERG_REPO="${ICEBERG_REPO:-$HOME/Desktop/work/iceberg}"
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+OUT_DIR="${OUT_DIR:-$SCRIPT_DIR/build/theta-fixtures}"
+TESTDATA_DIR="$(cd "$SCRIPT_DIR/../.." && pwd)/table/stats/testdata"
+
+# Versions Iceberg-Java pins (gradle/libs.versions.toml) and the memory
+# release datasketches-java depends on.
+DATASKETCHES_VERSION="$(sed -n 's/^datasketches = "\(.*\)"/\1/p'
"$ICEBERG_REPO/gradle/libs.versions.toml")"
+DATASKETCHES_MEMORY_VERSION="${DATASKETCHES_MEMORY_VERSION:-3.0.2}"
+
+resolve_jar() {
+ local pattern="$1"
+ local found
+ found="$(find "$ICEBERG_REPO" -path "*/build/libs/*" -name "$pattern" \
+ ! -name "*-tests.jar" ! -name "*-sources.jar" ! -name "*-javadoc.jar" !
-name "*-empty.jar" \
+ -print -quit 2>/dev/null)"
+ if [[ -z "$found" ]]; then
+ echo "ERROR: could not find $pattern under $ICEBERG_REPO/*/build/libs/" >&2
+ echo " Build it once: (cd $ICEBERG_REPO && ./gradlew
:iceberg-api:jar :iceberg-bundled-guava:jar)" >&2
+ exit 1
+ fi
+ echo "$found"
+}
+
+resolve_dep() {
+ local artifact="$1"
+ local version="$2"
+ local found
+ found="$(find
"$HOME/.gradle/caches/modules-2/files-2.1/org.apache.datasketches/$artifact/$version"
\
+ -name "${artifact}-${version}.jar" -print -quit 2>/dev/null)"
+ if [[ -z "$found" ]]; then
+ echo "ERROR: could not find ${artifact}-${version}.jar in the Gradle
cache" >&2
+ echo " Resolve it once: (cd $ICEBERG_REPO && ./gradlew
:iceberg-spark:iceberg-spark-4.1_2.13:dependencies)" >&2
+ exit 1
+ fi
+ echo "$found"
+}
+
+ICEBERG_API_JAR="$(resolve_jar 'iceberg-api-*.jar')"
+ICEBERG_GUAVA_JAR="$(resolve_jar 'iceberg-bundled-guava-*.jar')"
+DATASKETCHES_JAR="$(resolve_dep datasketches-java "$DATASKETCHES_VERSION")"
+MEMORY_JAR="$(resolve_dep datasketches-memory "$DATASKETCHES_MEMORY_VERSION")"
+
+CP="$ICEBERG_API_JAR:$ICEBERG_GUAVA_JAR:$DATASKETCHES_JAR:$MEMORY_JAR"
+
+CLASSES_DIR="$OUT_DIR/classes"
+mkdir -p "$CLASSES_DIR"
+
+echo "[generate.sh] compiling against:"
+echo " api: $ICEBERG_API_JAR"
+echo " guava: $ICEBERG_GUAVA_JAR"
+echo " datasketches: $DATASKETCHES_JAR"
+echo " memory: $MEMORY_JAR"
+
+javac -d "$CLASSES_DIR" -cp "$CP" "$SCRIPT_DIR/GenerateThetaFixtures.java"
+
+echo "[generate.sh] running"
+java -cp "$CLASSES_DIR:$CP" \
+ -Dtheta.fixtures.outputDir="$OUT_DIR" \
+ GenerateThetaFixtures
+
+echo "[generate.sh] mirroring into $TESTDATA_DIR"
+cp "$OUT_DIR"/*.sk "$TESTDATA_DIR/"
diff --git a/go.mod b/go.mod
index 5e1bd51d4..4485cd500 100644
--- a/go.mod
+++ b/go.mod
@@ -27,6 +27,7 @@ require (
github.com/RoaringBitmap/roaring/v2 v2.30.0
github.com/alexflint/go-arg v1.6.1
github.com/apache/arrow-go/v18 v18.8.0
+ github.com/apache/datasketches-go v0.2.0
github.com/aws/aws-sdk-go-v2 v1.47.4
github.com/aws/aws-sdk-go-v2/config v1.33.10
github.com/aws/aws-sdk-go-v2/credentials v1.20.10
diff --git a/go.sum b/go.sum
index 233132a12..ce9219015 100644
--- a/go.sum
+++ b/go.sum
@@ -76,6 +76,8 @@ github.com/antlr4-go/antlr/v4 v4.13.1
h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYW
github.com/antlr4-go/antlr/v4 v4.13.1/go.mod
h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
github.com/apache/arrow-go/v18 v18.8.0
h1:BLOzbPv7bxMPgXPacAg6HQjnxupYsZzC4tf+FkqPU/M=
github.com/apache/arrow-go/v18 v18.8.0/go.mod
h1:uJCFfCwq0KsxCmsCfQg4ft+LsW+iHYzAXiSDh5ug/8U=
+github.com/apache/datasketches-go v0.2.0
h1:whPblZiipxvN1+tiU3t4B8wnrIZRMmTN6oJyzhNxApA=
+github.com/apache/datasketches-go v0.2.0/go.mod
h1:2uFqFMsu21griPfN2rrK47rYO6XnJs4x+GvkrDwRbt8=
github.com/apache/thrift v0.24.0
h1:zy31L1a49QTNB2bG1BBfMXol3yJrTH975G3pPubQVLQ=
github.com/apache/thrift v0.24.0/go.mod
h1:zPt6WxgvTOM6hF92y8C+MkEM5LMxZuk4JcQOiU4Esvs=
github.com/aws/aws-sdk-go-v2 v1.47.4
h1:bTvk6UdSSSvaQsfLVj1JRDEEqj/PDRwGf1KanaRzau0=
diff --git a/table/stats/sketch.go b/table/stats/sketch.go
new file mode 100644
index 000000000..4b1616120
--- /dev/null
+++ b/table/stats/sketch.go
@@ -0,0 +1,241 @@
+// 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 stats computes Apache Iceberg table statistics.
+//
+// This file provides the number-of-distinct-values (NDV) primitive: a thin
+// wrapper over an Apache DataSketches Theta sketch that produces
+// apache-datasketches-theta-v1 statistics blobs mutually readable and
mergeable
+// with the ones written by Iceberg-Java's Spark ComputeTableStats action.
+// Interoperability rests on settings this package pins to match Iceberg-Java:
+// the hash seed (9001), the log2 nominal entries (lgK = 12, i.e. 4096
entries),
+// and the serialized form (an uncompressed, ordered compact sketch).
+//
+// Iceberg-Java builds one ALPHA-family update sketch per Spark partition and
+// merges the partials through a Theta union, so the sketch it stores is
always a
+// union result trimmed to 4096 entries. datasketches-go provides only the
+// QuickSelect update family, so this package mirrors the merge instead: Sketch
+// and Union both produce their estimate and serialization through a union with
+// the same settings. The decoded sketch (theta and retained hashes) is
therefore
+// identical to Iceberg-Java's whenever each per-partition sketch holds at most
+// 4096 distinct values, including when the merged result holds more and is
+// trimmed. Once a single sketch goes past 4096 distinct values, the ALPHA and
+// QuickSelect families keep different samples, so the estimates agree only
+// within the sketch's error bounds.
+//
+// Callers must feed values using Iceberg's single-value serialization (the
same
+// byte encoding used for manifest bounds: little-endian integers and floats,
+// UTF-8 strings with no length prefix, etc.); the sketch hashes those exact
+// bytes, so matching the encoding is what makes NDV estimates agree across
+// engines. Null values must not be added.
+package stats
+
+import (
+ "fmt"
+
+ "github.com/apache/datasketches-go/theta"
+)
+
+const (
+ // defaultSeed is the hash seed embedded in every serialized sketch. It
must
+ // match Iceberg-Java's ThetaUtil.DEFAULT_UPDATE_SEED (and the
DataSketches
+ // default) so sketches can be merged and read across implementations.
+ defaultSeed = theta.DefaultSeed
+
+ // defaultLgK is the log2 of the nominal number of entries.
Iceberg-Java's
+ // NDV action uses DataSketches' default of 4096 nominal entries (2^12)
for
+ // both its update sketches and its union, so we pin lgK = 12.
+ defaultLgK = uint8(12)
+)
+
+// Sketch accumulates values into a Theta sketch to estimate the number of
+// distinct values (NDV) of a column. It is not safe for concurrent use.
+//
+// Estimate, NDV and Serialize each build a new union over the retained
+// entries, so call them once after all values are added rather than per value.
+type Sketch struct {
+ us *theta.QuickSelectUpdateSketch
+}
+
+// NewSketch returns an empty Sketch configured to match Iceberg-Java's NDV
+// sketches (seed 9001, lgK 12).
+func NewSketch() *Sketch {
+ us, err := theta.NewQuickSelectUpdateSketch(
+ theta.WithUpdateSketchLgK(defaultLgK),
+ theta.WithUpdateSketchSeed(defaultSeed),
+ )
+ mustNotFail(err)
+
+ return &Sketch{us: us}
+}
+
+// Add feeds one non-null value into the sketch. value must be the Iceberg
+// single-value serialization of the column value; the sketch hashes these
+// bytes directly.
+//
+// An empty value (such as the empty string or a zero-length binary) is
+// ignored and not counted as a distinct value. This matches Iceberg-Java,
+// whose DataSketches UpdateSketch.update(ByteBuffer) rejects empty input.
+func (s *Sketch) Add(value []byte) {
+ if len(value) == 0 {
+ return
+ }
+
+ // The only errors UpdateBytes reports are a duplicate value, a hash at
or
+ // above the current sampling threshold, and a zero hash. Each is a
normal
+ // outcome that leaves the sketch unchanged, and DataSketches-Java
skips the
+ // same cases silently.
+ _ = s.us.UpdateBytes(value)
+}
+
+// Estimate returns the estimated number of distinct values added so far.
+func (s *Sketch) Estimate() float64 {
+ return s.result().Estimate()
+}
+
+// NDV returns Estimate truncated to an int64, matching how Iceberg-Java
records
+// the "ndv" blob property (String.valueOf((long) sketch.getEstimate())).
+func (s *Sketch) NDV() int64 {
+ return int64(s.Estimate())
+}
+
+// Serialize returns the compact, ordered, uncompressed serialization of the
+// sketch: the byte form stored inside an apache-datasketches-theta-v1 Puffin
+// blob (before any Puffin-level compression). The bytes are not identical to
+// Iceberg-Java's (DataSketches-Java sets a few preamble fields differently),
+// but each implementation decodes the other's to the same sketch.
+func (s *Sketch) Serialize() []byte {
+ return marshal(s.result())
+}
+
+// result passes the sketch through a union, as Iceberg-Java does with every
+// sketch it stores, so the result is trimmed to the nominal entries.
+func (s *Sketch) result() *theta.CompactSketch {
+ u := NewUnion()
+ u.Add(s)
+
+ return u.result()
+}
+
+// Union merges Theta sketches into a single sketch, used to combine per-file
or
+// per-partition sketches into one column-level NDV estimate. Merging requires
+// the same seed on every input, which this package guarantees. Union is not
+// safe for concurrent use.
+type Union struct {
+ u *theta.Union
+}
+
+// NewUnion returns an empty Union configured to match Iceberg-Java's NDV union
+// (seed 9001, lgK 12).
+func NewUnion() *Union {
+ u, err := theta.NewUnion(
+ theta.WithUnionLgK(defaultLgK),
+ theta.WithUnionSeed(defaultSeed),
+ )
+ mustNotFail(err)
+
+ return &Union{u: u}
+}
+
+// Add merges s into the union.
+func (u *Union) Add(s *Sketch) {
+ // Update fails only on a seed mismatch, which cannot happen between
+ // sketches this package configures.
+ mustNotFail(u.u.Update(s.us))
+}
+
+// AddBytes decodes a serialized sketch (as produced by Sketch.Serialize or by
+// another Iceberg implementation) and merges it into the union.
+func (u *Union) AddBytes(serialized []byte) error {
+ cs, err := theta.Decode(serialized, defaultSeed)
+ if err != nil {
+ return fmt.Errorf("stats: decode theta sketch: %w", err)
+ }
+ if err := u.u.Update(cs); err != nil {
+ return fmt.Errorf("stats: union theta sketch: %w", err)
+ }
+
+ return nil
+}
+
+// Estimate returns the estimated number of distinct values across all merged
+// sketches.
+func (u *Union) Estimate() float64 {
+ return u.result().Estimate()
+}
+
+// NDV returns Estimate truncated to an int64.
+func (u *Union) NDV() int64 {
+ return int64(u.Estimate())
+}
+
+// Serialize returns the compact, ordered, uncompressed serialization of the
+// merged sketch, in the same byte form as Sketch.Serialize.
+func (u *Union) Serialize() []byte {
+ return marshal(u.result())
+}
+
+func (u *Union) result() *theta.CompactSketch {
+ // Result fails only if the seed hash cannot be computed, which cannot
+ // happen for the fixed seed this package uses.
+ cs, err := u.u.Result(true)
+ mustNotFail(err)
+
+ return cs
+}
+
+// EstimateFromBytes decodes a serialized Theta sketch and returns its
estimated
+// number of distinct values. It reads sketches written by any Iceberg
+// implementation, including the compressed serialization versions produced by
+// newer DataSketches releases.
+func EstimateFromBytes(serialized []byte) (float64, error) {
+ cs, err := theta.Decode(serialized, defaultSeed)
+ if err != nil {
+ return 0, fmt.Errorf("stats: decode theta sketch: %w", err)
+ }
+
+ return cs.Estimate(), nil
+}
+
+// NDVFromBytes decodes a serialized Theta sketch and returns its estimate
+// truncated to an int64.
+func NDVFromBytes(serialized []byte) (int64, error) {
+ est, err := EstimateFromBytes(serialized)
+ if err != nil {
+ return 0, err
+ }
+
+ return int64(est), nil
+}
+
+func marshal(cs *theta.CompactSketch) []byte {
+ // Uncompressed encoding into memory fails only if the writer does, and
a
+ // bytes.Buffer never does.
+ b, err := cs.MarshalBinary()
+ mustNotFail(err)
+
+ return b
+}
+
+// mustNotFail panics on errors that the pinned settings rule out. Reaching it
+// means datasketches-go changed behavior, not that the caller did anything
+// wrong.
+func mustNotFail(err error) {
+ if err != nil {
+ panic(fmt.Sprintf("stats: unexpected theta sketch error: %v",
err))
+ }
+}
diff --git a/table/stats/sketch_test.go b/table/stats/sketch_test.go
new file mode 100644
index 000000000..d06cae904
--- /dev/null
+++ b/table/stats/sketch_test.go
@@ -0,0 +1,456 @@
+// 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 stats_test
+
+import (
+ "encoding/binary"
+ "fmt"
+ "math"
+ "os"
+ "path/filepath"
+ "slices"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/apache/datasketches-go/theta"
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table/stats"
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// leInt64 encodes v the way Iceberg serializes a LONG value: 8 bytes,
+// little-endian.
+func leInt64(v int64) []byte {
+ b := make([]byte, 8)
+ binary.LittleEndian.PutUint64(b, uint64(v))
+
+ return b
+}
+
+func TestSketchExactModeIsExact(t *testing.T) {
+ s := stats.NewSketch()
+
+ const n = 100
+ for i := range int64(n) {
+ s.Add(leInt64(i))
+ }
+
+ // Below the nominal entry count the sketch retains every distinct
value, so
+ // the estimate is exact.
+ assert.Equal(t, float64(n), s.Estimate())
+ assert.Equal(t, int64(n), s.NDV())
+}
+
+func TestSketchIgnoresDuplicates(t *testing.T) {
+ s := stats.NewSketch()
+
+ for range 1000 {
+ s.Add(leInt64(42))
+ }
+
+ assert.Equal(t, int64(1), s.NDV())
+}
+
+func TestSketchEmpty(t *testing.T) {
+ s := stats.NewSketch()
+
+ assert.Equal(t, 0.0, s.Estimate())
+ assert.Equal(t, int64(0), s.NDV())
+
+ ndv, err := stats.NDVFromBytes(s.Serialize())
+ require.NoError(t, err)
+ assert.Equal(t, int64(0), ndv)
+}
+
+// TestSketchSkipsEmptyValues checks that empty values are not counted, as in
+// Iceberg-Java: a string column with "", "a", "b" has an NDV of 2.
+func TestSketchSkipsEmptyValues(t *testing.T) {
+ s := stats.NewSketch()
+
+ s.Add([]byte(""))
+ s.Add([]byte("a"))
+ s.Add([]byte("b"))
+ s.Add(nil)
+ s.Add([]byte{})
+
+ assert.Equal(t, int64(2), s.NDV())
+
+ onlyEmpty := stats.NewSketch()
+ onlyEmpty.Add([]byte(""))
+ assert.Equal(t, int64(0), onlyEmpty.NDV())
+}
+
+func TestSketchDistinctByteEncoding(t *testing.T) {
+ s := stats.NewSketch()
+
+ s.Add([]byte("apple"))
+ s.Add([]byte("apple"))
+ s.Add([]byte("banana"))
+
+ assert.Equal(t, int64(2), s.NDV())
+}
+
+// TestSketchTrimsToNominalEntries checks that a sketch past 4096 distinct
values
+// is trimmed to 4096 retained entries, as Iceberg-Java's union does with every
+// sketch it stores, so it reports an estimate rather than an exact count.
+func TestSketchTrimsToNominalEntries(t *testing.T) {
+ s := stats.NewSketch()
+ for i := range int64(6000) {
+ s.Add(leInt64(i))
+ }
+
+ cs, err := theta.Decode(s.Serialize(), theta.DefaultSeed)
+ require.NoError(t, err)
+ assert.Equal(t, uint32(4096), cs.NumRetained())
+ assert.True(t, cs.IsEstimationMode())
+ assert.Equal(t, cs.Estimate(), s.Estimate())
+ assert.NotEqual(t, int64(6000), s.NDV())
+}
+
+func TestSerializeRoundTrip(t *testing.T) {
+ s := stats.NewSketch()
+ for i := range int64(5000) {
+ s.Add(leInt64(i))
+ }
+
+ est, err := stats.EstimateFromBytes(s.Serialize())
+ require.NoError(t, err)
+ assert.Equal(t, s.Estimate(), est)
+}
+
+func TestUnionSketches(t *testing.T) {
+ a := stats.NewSketch()
+ b := stats.NewSketch()
+
+ // [0,60) and [40,100): 100 distinct across the two, 20 overlapping.
+ for i := range int64(60) {
+ a.Add(leInt64(i))
+ }
+ for i := int64(40); i < 100; i++ {
+ b.Add(leInt64(i))
+ }
+
+ u := stats.NewUnion()
+ u.Add(a)
+ u.Add(b)
+
+ assert.Equal(t, int64(100), u.NDV())
+}
+
+func TestUnionAddBytes(t *testing.T) {
+ a := stats.NewSketch()
+ for i := range int64(60) {
+ a.Add(leInt64(i))
+ }
+
+ b := stats.NewSketch()
+ for i := int64(40); i < 100; i++ {
+ b.Add(leInt64(i))
+ }
+
+ u := stats.NewUnion()
+ require.NoError(t, u.AddBytes(a.Serialize()))
+ u.Add(b)
+
+ assert.Equal(t, int64(100), u.NDV())
+
+ got, err := stats.NDVFromBytes(u.Serialize())
+ require.NoError(t, err)
+ assert.Equal(t, int64(100), got)
+}
+
+func TestUnionAddBytesRejectsInvalidInput(t *testing.T) {
+ u := stats.NewUnion()
+ assert.Error(t, u.AddBytes([]byte{1, 2, 3}))
+
+ _, err := stats.EstimateFromBytes([]byte{1, 2, 3})
+ assert.Error(t, err)
+}
+
+// javaFixture describes one sketch in testdata produced by
+// dev/theta-fixtures/GenerateThetaFixtures.java. partitions mirrors the
+// generator's input: one slice of rows per Spark partition.
+type javaFixture struct {
+ name string
+ ndv int64
+ partitions [][]row
+}
+
+// row is one column value, encoded with Iceberg's single-value serialization.
+// Null is explicit because an encoded empty value (such as "") may be a nil
+// slice, and empty values must still reach Sketch.Add.
+type row struct {
+ value []byte
+ null bool
+}
+
+var nullRow = row{null: true}
+
+func lit[T iceberg.LiteralType](t *testing.T, v T) []byte {
+ b, err := iceberg.NewLiteral(v).MarshalBinary()
+ require.NoError(t, err)
+
+ return b
+}
+
+// dealRows mirrors GenerateThetaFixtures.write: rows 0..numRows-1 are dealt
+// round-robin across numPartitions partitions.
+func dealRows(numPartitions, numRows int, value func(i int) row) [][]row {
+ partitions := make([][]row, numPartitions)
+ for i := range numRows {
+ p := i % numPartitions
+ partitions[p] = append(partitions[p], value(i))
+ }
+
+ return partitions
+}
+
+func nullEvery(i, every int, value func() []byte) row {
+ if i%every == every-1 {
+ return nullRow
+ }
+
+ return row{value: value()}
+}
+
+// longRanges mirrors GenerateThetaFixtures.range: partition p holds count
+// consecutive longs starting at p*step.
+func longRanges(t *testing.T, numPartitions int, step int64, count int)
[][]row {
+ partitions := make([][]row, numPartitions)
+ for p := range numPartitions {
+ start := int64(p) * step
+ for i := range int64(count) {
+ partitions[p] = append(partitions[p], row{value: lit(t,
start+i)})
+ }
+ }
+
+ return partitions
+}
+
+func javaFixtures(t *testing.T) []javaFixture {
+ const microsBase = int64(1_600_000_000_000_000)
+
+ return []javaFixture{
+ {name: "null_only", ndv: 0, partitions: dealRows(1, 3,
func(int) row { return nullRow })},
+ {name: "boolean", ndv: 2, partitions: dealRows(3, 100, func(i
int) row {
+ return nullEvery(i, 7, func() []byte { return lit(t,
i%2 == 0) })
+ })},
+ {name: "int", ndv: 582, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte { return lit(t,
int32((i%600)*7-2000)) })
+ })},
+ {name: "long", ndv: 582, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte { return lit(t,
int64(i%600)*1_000_003-5_000_000_000) })
+ })},
+ {name: "float", ndv: 582, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte {
+ switch i % 600 {
+ case 0:
+ return lit(t, float32(0))
+ case 1:
+ return lit(t, float32(math.Copysign(0,
-1)))
+ default:
+ return lit(t, float32(i%600)*0.5+1)
+ }
+ })
+ })},
+ {name: "double", ndv: 582, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte { return lit(t,
float64(i%600)/3.0) })
+ })},
+ {name: "date", ndv: 400, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte { return lit(t,
iceberg.Date(18000+i%400)) })
+ })},
+ {name: "timestamp", ndv: 828, partitions: dealRows(3, 1000,
func(i int) row {
+ return nullEvery(i, 11, func() []byte {
+ return lit(t,
iceberg.Timestamp(microsBase+int64(i%900)*1_000_000))
+ })
+ })},
+ {name: "string", ndv: 499, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte {
+ switch i % 500 {
+ case 0:
+ return lit(t, "")
+ case 1:
+ return lit(t, "é")
+ default:
+ return lit(t, fmt.Sprintf("value-%d",
i%500))
+ }
+ })
+ })},
+ {name: "binary", ndv: 299, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte {
+ if i%300 == 0 {
+ return lit(t, []byte{})
+ }
+
+ return lit(t, []byte{byte(i % 300), byte(i %
300 / 7)})
+ })
+ })},
+ {name: "decimal", ndv: 582, partitions: dealRows(3, 1000,
func(i int) row {
+ return nullEvery(i, 11, func() []byte {
+ unscaled := int64(i%600-300) * 12345
+
+ return lit(t, iceberg.Decimal{Val:
decimal128.FromI64(unscaled), Scale: 2})
+ })
+ })},
+ {name: "uuid", ndv: 582, partitions: dealRows(3, 1000, func(i
int) row {
+ return nullEvery(i, 11, func() []byte {
+ var u uuid.UUID
+ binary.BigEndian.PutUint64(u[:8], uint64(i%600))
+ binary.BigEndian.PutUint64(u[8:],
uint64(i%600)*31+7)
+
+ return lit(t, u)
+ })
+ })},
+ {name: "string_empty_only", ndv: 0, partitions: dealRows(2, 10,
func(int) row { return row{value: lit(t, "")} })},
+ {name: "merge_exact_partitions", ndv: 5102, partitions:
longRanges(t, 4, 1200, 1500)},
+ {name: "single_partition_at_k", ndv: 4096, partitions:
longRanges(t, 1, 0, 4096)},
+ }
+}
+
+// buildUnion mirrors the Spark aggregation in GenerateThetaFixtures: one
+// sketch per partition, nulls skipped, partials merged through a union.
+func buildUnion(partitions [][]row) *stats.Union {
+ u := stats.NewUnion()
+ for _, rows := range partitions {
+ s := stats.NewSketch()
+ for _, r := range rows {
+ if !r.null {
+ s.Add(r.value)
+ }
+ }
+ u.Add(s)
+ }
+
+ return u
+}
+
+func readFixture(t *testing.T, name string) []byte {
+ t.Helper()
+ b, err := os.ReadFile(filepath.Join("testdata", name+".sk"))
+ require.NoError(t, err)
+
+ return b
+}
+
+func decode(t *testing.T, b []byte) *theta.CompactSketch {
+ t.Helper()
+ cs, err := theta.Decode(b, theta.DefaultSeed)
+ require.NoError(t, err)
+
+ return cs
+}
+
+// assertSameSketch compares decoded state rather than raw bytes:
+// DataSketches-Java writes a few preamble fields differently (for example p
+// and the single-item flag), but equal emptiness, theta and retained hashes
+// mean the sketches are identical.
+func assertSameSketch(t *testing.T, java, ours *theta.CompactSketch) {
+ t.Helper()
+ assert.Equal(t, java.IsEmpty(), ours.IsEmpty(), "empty")
+ assert.Equal(t, java.Theta64(), ours.Theta64(), "theta")
+ assert.Equal(t, java.NumRetained(), ours.NumRetained(), "retained")
+ assert.Equal(t, slices.Collect(java.All()), slices.Collect(ours.All()),
"retained hashes")
+}
+
+// TestJavaFixtureParity rebuilds each Iceberg-Java fixture from the same
+// values and checks that the merged sketch decodes to the identical sketch and
+// NDV. Values are encoded with iceberg-go's literal serialization, so the test
+// also pins that encoding to Java's Conversions.toByteBuffer for every type.
+func TestJavaFixtureParity(t *testing.T) {
+ for _, fx := range javaFixtures(t) {
+ t.Run(fx.name, func(t *testing.T) {
+ javaBytes := readFixture(t, fx.name)
+ javaNDV, err := stats.NDVFromBytes(javaBytes)
+ require.NoError(t, err)
+ require.Equal(t, fx.ndv, javaNDV, "fixture ndv")
+
+ u := buildUnion(fx.partitions)
+
+ assert.Equal(t, javaNDV, u.NDV())
+ assertSameSketch(t, decode(t, javaBytes), decode(t,
u.Serialize()))
+
+ if len(fx.partitions) == 1 {
+ s := stats.NewSketch()
+ for _, r := range fx.partitions[0] {
+ if !r.null {
+ s.Add(r.value)
+ }
+ }
+ assert.Equal(t, javaNDV, s.NDV())
+ assertSameSketch(t, decode(t, javaBytes),
decode(t, s.Serialize()))
+ }
+ })
+ }
+}
+
+// TestJavaFixtureReadAndMerge checks that sketches Iceberg-Java wrote can be
+// read and merged with Go-built ones: a union of the Java fixture and the same
+// values built in Go is unchanged.
+func TestJavaFixtureReadAndMerge(t *testing.T) {
+ javaBytes := readFixture(t, "merge_exact_partitions")
+
+ s := stats.NewSketch()
+ for i := range int64(5100) {
+ s.Add(leInt64(i))
+ }
+
+ u := stats.NewUnion()
+ require.NoError(t, u.AddBytes(javaBytes))
+ u.Add(s)
+
+ assertSameSketch(t, decode(t, javaBytes), decode(t, u.Serialize()))
+}
+
+// TestJavaEstimationModeWithinBounds covers fixtures where a single
+// per-partition sketch goes past 4096 distinct values. Iceberg-Java's ALPHA
+// sketch and our QuickSelect sketch then keep different samples, so the
+// estimates are not identical; both must stay within the other's error bounds.
+func TestJavaEstimationModeWithinBounds(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ partitions [][]row
+ distinct float64
+ }{
+ {"single_partition_estimation", longRanges(t, 1, 0, 6000),
6000},
+ {"multi_partition_estimation", longRanges(t, 3, 3000, 6000),
12000},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ java := decode(t, readFixture(t, tc.name))
+
+ ours := decode(t, buildUnion(tc.partitions).Serialize())
+
+ for _, cs := range []*theta.CompactSketch{java, ours} {
+ lb, err := cs.LowerBound(3)
+ require.NoError(t, err)
+ ub, err := cs.UpperBound(3)
+ require.NoError(t, err)
+ assert.True(t, lb <= tc.distinct && tc.distinct
<= ub,
+ "true count %v outside [%v, %v]",
tc.distinct, lb, ub)
+ }
+
+ lb, err := java.LowerBound(3)
+ require.NoError(t, err)
+ ub, err := java.UpperBound(3)
+ require.NoError(t, err)
+ assert.True(t, lb <= ours.Estimate() && ours.Estimate()
<= ub,
+ "our estimate %v outside Java's bounds [%v,
%v]", ours.Estimate(), lb, ub)
+ })
+ }
+}
diff --git a/table/stats/testdata/README.md b/table/stats/testdata/README.md
new file mode 100644
index 000000000..d2edb983f
--- /dev/null
+++ b/table/stats/testdata/README.md
@@ -0,0 +1,56 @@
+<!--
+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.
+-->
+
+# Theta sketch test fixtures
+
+The `.sk` files are Theta sketches produced the way Iceberg-Java's Spark
+`ComputeTableStats` action produces the `apache-datasketches-theta-v1` blobs it
+writes to Puffin statistics files. They are generated by
+`dev/theta-fixtures/GenerateThetaFixtures.java` (see its README to regenerate)
+and consumed by `table/stats/sketch_test.go`.
+
+The generator replays Spark's `ThetaSketchAgg` with the DataSketches-Java
+release Iceberg pins (6.2.0): an ALPHA update sketch per partition, non-null
+values hashed from `Conversions.toByteBuffer`, partials merged through a union,
+and the result stored as an uncompressed compact sketch (serialization
+version 3). The Go tests rebuild each sketch from the same values, encoded with
+iceberg-go's literal serialization, and compare decoded state: emptiness,
+theta, and retained hashes. Raw bytes are not compared, because
+DataSketches-Java sets a few preamble fields differently.
+
+| file | input | Java ndv |
+| --- | --- | --- |
+| `null_only.sk` | only nulls | 0 |
+| `boolean.sk` … `uuid.sk` | one fixture per primitive type, 3 partitions,
with duplicates and nulls | see test |
+| `string.sk`, `binary.sk` | include empty values, which Java skips | 499, 299
|
+| `string_empty_only.sk` | only empty strings | 0 |
+| `merge_exact_partitions.sk` | 4 partitions of 1500 longs each, 5100 distinct
overall | 5102 |
+| `single_partition_at_k.sk` | 1 partition of 4096 distinct longs | 4096 |
+| `single_partition_estimation.sk` | 1 partition of 6000 distinct longs | 6007
|
+| `multi_partition_estimation.sk` | 3 overlapping partitions of 6000 longs,
12000 distinct overall | 11973 |
+
+Every fixture up to `single_partition_at_k.sk` must decode identically to the
+sketch iceberg-go builds. In the two estimation fixtures a single partition
+holds more than 4096 distinct values, where the ALPHA and QuickSelect families
+keep different samples; those are checked against each other's error bounds.
+
+UUID values are hashed from their 16-byte single-value serialization, as the
+Puffin spec requires. Spark reads Iceberg `uuid` columns as strings, so
+sketches written by the Spark action for UUID columns hash the 36-character
+text form instead and will not match.
diff --git a/table/stats/testdata/binary.sk b/table/stats/testdata/binary.sk
new file mode 100644
index 000000000..a75cb8786
Binary files /dev/null and b/table/stats/testdata/binary.sk differ
diff --git a/table/stats/testdata/boolean.sk b/table/stats/testdata/boolean.sk
new file mode 100644
index 000000000..c7ff0f645
Binary files /dev/null and b/table/stats/testdata/boolean.sk differ
diff --git a/table/stats/testdata/date.sk b/table/stats/testdata/date.sk
new file mode 100644
index 000000000..3dbac41ea
Binary files /dev/null and b/table/stats/testdata/date.sk differ
diff --git a/table/stats/testdata/decimal.sk b/table/stats/testdata/decimal.sk
new file mode 100644
index 000000000..8e46e4291
Binary files /dev/null and b/table/stats/testdata/decimal.sk differ
diff --git a/table/stats/testdata/double.sk b/table/stats/testdata/double.sk
new file mode 100644
index 000000000..f4a224a18
Binary files /dev/null and b/table/stats/testdata/double.sk differ
diff --git a/table/stats/testdata/float.sk b/table/stats/testdata/float.sk
new file mode 100644
index 000000000..676af1fb4
Binary files /dev/null and b/table/stats/testdata/float.sk differ
diff --git a/table/stats/testdata/int.sk b/table/stats/testdata/int.sk
new file mode 100644
index 000000000..b8052add4
Binary files /dev/null and b/table/stats/testdata/int.sk differ
diff --git a/table/stats/testdata/long.sk b/table/stats/testdata/long.sk
new file mode 100644
index 000000000..8fde1b2ef
Binary files /dev/null and b/table/stats/testdata/long.sk differ
diff --git a/table/stats/testdata/merge_exact_partitions.sk
b/table/stats/testdata/merge_exact_partitions.sk
new file mode 100644
index 000000000..755030935
Binary files /dev/null and b/table/stats/testdata/merge_exact_partitions.sk
differ
diff --git a/table/stats/testdata/multi_partition_estimation.sk
b/table/stats/testdata/multi_partition_estimation.sk
new file mode 100644
index 000000000..eced37c33
Binary files /dev/null and b/table/stats/testdata/multi_partition_estimation.sk
differ
diff --git a/table/stats/testdata/null_only.sk
b/table/stats/testdata/null_only.sk
new file mode 100644
index 000000000..f6c647f8a
Binary files /dev/null and b/table/stats/testdata/null_only.sk differ
diff --git a/table/stats/testdata/single_partition_at_k.sk
b/table/stats/testdata/single_partition_at_k.sk
new file mode 100644
index 000000000..b7070f8ab
Binary files /dev/null and b/table/stats/testdata/single_partition_at_k.sk
differ
diff --git a/table/stats/testdata/single_partition_estimation.sk
b/table/stats/testdata/single_partition_estimation.sk
new file mode 100644
index 000000000..1a005bebf
Binary files /dev/null and
b/table/stats/testdata/single_partition_estimation.sk differ
diff --git a/table/stats/testdata/string.sk b/table/stats/testdata/string.sk
new file mode 100644
index 000000000..0dd8503cb
Binary files /dev/null and b/table/stats/testdata/string.sk differ
diff --git a/table/stats/testdata/string_empty_only.sk
b/table/stats/testdata/string_empty_only.sk
new file mode 100644
index 000000000..f6c647f8a
Binary files /dev/null and b/table/stats/testdata/string_empty_only.sk differ
diff --git a/table/stats/testdata/timestamp.sk
b/table/stats/testdata/timestamp.sk
new file mode 100644
index 000000000..dbf4ccae2
Binary files /dev/null and b/table/stats/testdata/timestamp.sk differ
diff --git a/table/stats/testdata/uuid.sk b/table/stats/testdata/uuid.sk
new file mode 100644
index 000000000..e2e5bf79b
Binary files /dev/null and b/table/stats/testdata/uuid.sk differ
