laskoviymishka commented on code in PR #2444: URL: https://github.com/apache/iceberg-go/pull/2444#discussion_r4536869222
########## table/stats_ndv.go: ########## @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table/stats" +) + +// NDVResult is the number-of-distinct-values sketch computed for one column. +type NDVResult struct { + // FieldID is the Iceberg field ID of the column. + FieldID int + // NDV is the estimated number of distinct non-null values. + NDV int64 + // Sketch is the serialized DataSketches Theta sketch, i.e. the body of an + // apache-datasketches-theta-v1 statistics blob. + Sketch []byte +} + +// ComputeNDVs scans a snapshot and computes a DataSketches Theta sketch per +// top-level primitive column, returning the estimated number of distinct values +// keyed by field ID. +// +// When snapshotID is 0 the current snapshot is used. When columns is empty every +// top-level primitive column is analyzed, matching Iceberg-Java's +// ComputeTableStats default. Naming an unknown column, a non-primitive column, +// or a column nested inside a struct is an error. +// +// Values are fed to the sketch using Iceberg's single-value serialization (the +// same encoding used for manifest bounds), so estimates are comparable with +// sketches produced by other Iceberg implementations. Null values are not +// counted. +func (t Table) ComputeNDVs(ctx context.Context, snapshotID int64, columns ...string) (map[int]NDVResult, error) { + opts := []ScanOption{WithSnapshotID(snapshotID)} + if len(columns) > 0 { Review Comment: Could the default path pass the top-level primitive names to `WithSelectedFields`? With no columns the scan stays on `"*"` ([projection](https://github.com/apache/iceberg-go/blob/5ec63084d3e328bf61cab878b99c117592947d7f/table/scanner.go#L734-L736)), so struct, list and map columns are read and decoded in full, then dropped by the filter below. On `{id long, s struct<a long>}` the scan decodes `s` with its values. Results are right, but wide nested tables pay that I/O and memory on every call. ########## table/stats_ndv.go: ########## @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table/stats" +) + +// NDVResult is the number-of-distinct-values sketch computed for one column. +type NDVResult struct { + // FieldID is the Iceberg field ID of the column. + FieldID int + // NDV is the estimated number of distinct non-null values. + NDV int64 + // Sketch is the serialized DataSketches Theta sketch, i.e. the body of an + // apache-datasketches-theta-v1 statistics blob. + Sketch []byte +} + +// ComputeNDVs scans a snapshot and computes a DataSketches Theta sketch per +// top-level primitive column, returning the estimated number of distinct values +// keyed by field ID. +// +// When snapshotID is 0 the current snapshot is used. When columns is empty every +// top-level primitive column is analyzed, matching Iceberg-Java's +// ComputeTableStats default. Naming an unknown column, a non-primitive column, +// or a column nested inside a struct is an error. +// +// Values are fed to the sketch using Iceberg's single-value serialization (the +// same encoding used for manifest bounds), so estimates are comparable with +// sketches produced by other Iceberg implementations. Null values are not +// counted. +func (t Table) ComputeNDVs(ctx context.Context, snapshotID int64, columns ...string) (map[int]NDVResult, error) { Review Comment: Should this return the snapshot it resolved along with the map? A Puffin writer for #1929 needs the snapshot ID and sequence number, which Java's blob carries ([`NDVSketchUtil`](https://github.com/apache/iceberg/blob/7f07cfb6ec4444706af66e8f9c7c5e83c1ff5071/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/NDVSketchUtil.java#L64-L75)), and with `0` the caller has to look it up again. A result struct holding the snapshot and the per-field map would do, but I'm open on the shape before apidiff locks it in. ########## table/stats_ndv.go: ########## @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table/stats" +) + +// NDVResult is the number-of-distinct-values sketch computed for one column. +type NDVResult struct { + // FieldID is the Iceberg field ID of the column. + FieldID int + // NDV is the estimated number of distinct non-null values. + NDV int64 + // Sketch is the serialized DataSketches Theta sketch, i.e. the body of an + // apache-datasketches-theta-v1 statistics blob. + Sketch []byte +} + +// ComputeNDVs scans a snapshot and computes a DataSketches Theta sketch per +// top-level primitive column, returning the estimated number of distinct values +// keyed by field ID. +// +// When snapshotID is 0 the current snapshot is used. When columns is empty every +// top-level primitive column is analyzed, matching Iceberg-Java's +// ComputeTableStats default. Naming an unknown column, a non-primitive column, +// or a column nested inside a struct is an error. +// +// Values are fed to the sketch using Iceberg's single-value serialization (the +// same encoding used for manifest bounds), so estimates are comparable with +// sketches produced by other Iceberg implementations. Null values are not +// counted. +func (t Table) ComputeNDVs(ctx context.Context, snapshotID int64, columns ...string) (map[int]NDVResult, error) { + opts := []ScanOption{WithSnapshotID(snapshotID)} + if len(columns) > 0 { + opts = append(opts, WithSelectedFields(columns...)) + } + scan := t.Scan(opts...) + + // Resolve the snapshot so an empty or missing target is a clear error rather + // than a silently empty result. + if _, err := scan.ResolveSnapshot(); err != nil { Review Comment: Could a nil snapshot return an empty map here, like Java's [`EMPTY_RESULT`](https://github.com/apache/iceberg/blob/7f07cfb6ec4444706af66e8f9c7c5e83c1ff5071/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ComputeTableStatsSparkAction.java#L88-L97)? On a table with no snapshot, [`ResolveSnapshot`](https://github.com/apache/iceberg-go/blob/5ec63084d3e328bf61cab878b99c117592947d7f/table/scanner.go#L678-L707) returns `(nil, nil)`, so the call succeeds with an 8-byte NDV=0 sketch per column. That's the silently empty result this comment says we avoid. A test on a fresh table would pin it. ########## table/stats_ndv_test.go: ########## @@ -0,0 +1,288 @@ +// 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 table_test + +import ( + "context" + "fmt" + "os" + "path/filepath" + "slices" + "strings" + "testing" + + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/datasketches-go/theta" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/catalog/sql" + "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/table/stats" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uptrace/bun/driver/sqliteshim" +) + +// createNDVTable creates a fresh sqlite-backed table named name with the given +// schema and appends one record batch built from the JSON rows. +func createNDVTable(t *testing.T, name string, sc *iceberg.Schema, jsonRows []string) (*table.Table, context.Context) { + t.Helper() + ctx := context.Background() + cat, err := catalog.Load(ctx, "default", iceberg.Properties{ + "uri": ":memory:", + "type": "sql", + sql.DriverKey: sqliteshim.ShimName, + sql.DialectKey: string(sql.SQLite), + "warehouse": "file://" + t.TempDir(), + }) + require.NoError(t, err) + + ident := table.Identifier{"default", name} + require.NoError(t, cat.CreateNamespace(ctx, catalog.NamespaceFromIdent(ident), nil)) + + tbl, err := cat.CreateTable(ctx, ident, sc) + require.NoError(t, err) + + arrowSchema, err := table.SchemaToArrowSchema(sc, nil, true, false) + require.NoError(t, err) + arrTable, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, + []string{"[" + strings.Join(jsonRows, ",") + "]"}) + require.NoError(t, err) + defer arrTable.Release() + + tbl, err = tbl.AppendTable(ctx, arrTable, arrTable.NumRows(), nil) + require.NoError(t, err) + + return tbl, ctx +} + +// ndvTestTable creates a fresh sqlite-backed table named tbl and appends rows: +// 100 distinct ids (1..100) and a category cycling through 3 distinct values. +func ndvTestTable(t *testing.T, name string) (*table.Table, context.Context) { + t.Helper() + sc := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "category", Type: iceberg.PrimitiveTypes.String, Required: true}, + ) + + categories := []string{"north", "south", "east"} + rows := make([]string, 0, 100) + for i := 1; i <= 100; i++ { + rows = append(rows, fmt.Sprintf(`{"id": %d, "category": %q}`, i, categories[i%len(categories)])) + } + + return createNDVTable(t, name, sc, rows) +} + +func TestComputeNDVsAllColumns(t *testing.T) { + tbl, ctx := ndvTestTable(t, "ndv_all") + + results, err := tbl.ComputeNDVs(ctx, 0) + require.NoError(t, err) + require.Len(t, results, 2) + + // Both cardinalities are well below the nominal entry count, so the sketch + // estimates are exact. + assert.Equal(t, int64(100), results[1].NDV) + assert.Equal(t, int64(3), results[2].NDV) + + // The serialized sketches carry the same estimate when read back. + for _, r := range results { + ndv, err := stats.NDVFromBytes(r.Sketch) + require.NoError(t, err) + assert.Equal(t, r.NDV, ndv) + } +} + +func TestComputeNDVsSelectedColumn(t *testing.T) { + tbl, ctx := ndvTestTable(t, "ndv_selected") + + results, err := tbl.ComputeNDVs(ctx, 0, "category") + require.NoError(t, err) + require.Len(t, results, 1) + assert.Equal(t, int64(3), results[2].NDV) +} + +func TestComputeNDVsExcludesUnselectedColumns(t *testing.T) { + tbl, ctx := ndvTestTable(t, "ndv_unselected") + + // Selecting only "category" yields exactly one sketch; the unselected "id" + // column is absent from the results. + results, err := tbl.ComputeNDVs(ctx, 0, "category") + require.NoError(t, err) + _, hasID := results[1] + assert.False(t, hasID) + _, hasCategory := results[2] + assert.True(t, hasCategory) +} + +func TestComputeNDVsUnknownColumnErrors(t *testing.T) { + tbl, ctx := ndvTestTable(t, "ndv_unknown") + + // A column that does not exist in the schema fails projection rather than + // being silently skipped. + _, err := tbl.ComputeNDVs(ctx, 0, "does_not_exist") + require.Error(t, err) +} + +func TestComputeNDVsPrimitiveTypes(t *testing.T) { Review Comment: Could we turn this into a table-driven matrix in this PR? Every test goes through [one `AppendTable` with snapshot `0`](https://github.com/apache/iceberg-go/blob/5ec63084d3e328bf61cab878b99c117592947d7f/table/stats_ndv_test.go#L42-L73), so nothing covers position or equality deletes, an older snapshot after a schema change, or several data files. The matrix could also add date, time, timestamp, timestamptz, decimal, uuid, binary and fixed. @zeroshade checked those by hand, and committed cases keep them pinned once the Arrow-buffer hashing follow-up replaces that path. ########## table/stats_ndv.go: ########## @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table/stats" +) + +// NDVResult is the number-of-distinct-values sketch computed for one column. +type NDVResult struct { + // FieldID is the Iceberg field ID of the column. + FieldID int + // NDV is the estimated number of distinct non-null values. + NDV int64 + // Sketch is the serialized DataSketches Theta sketch, i.e. the body of an + // apache-datasketches-theta-v1 statistics blob. + Sketch []byte +} + +// ComputeNDVs scans a snapshot and computes a DataSketches Theta sketch per +// top-level primitive column, returning the estimated number of distinct values +// keyed by field ID. +// +// When snapshotID is 0 the current snapshot is used. When columns is empty every +// top-level primitive column is analyzed, matching Iceberg-Java's +// ComputeTableStats default. Naming an unknown column, a non-primitive column, +// or a column nested inside a struct is an error. +// +// Values are fed to the sketch using Iceberg's single-value serialization (the +// same encoding used for manifest bounds), so estimates are comparable with +// sketches produced by other Iceberg implementations. Null values are not +// counted. +func (t Table) ComputeNDVs(ctx context.Context, snapshotID int64, columns ...string) (map[int]NDVResult, error) { + opts := []ScanOption{WithSnapshotID(snapshotID)} + if len(columns) > 0 { + opts = append(opts, WithSelectedFields(columns...)) + } + scan := t.Scan(opts...) + + // Resolve the snapshot so an empty or missing target is a clear error rather + // than a silently empty result. + if _, err := scan.ResolveSnapshot(); err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + + projected, err := scan.Projection() + if err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + + if err := validateNDVColumns(projected, columns); err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + + // Restrict to top-level primitive fields; nested fields are not sketched, + // matching the reference implementation. + primitives := make(map[string]iceberg.NestedField) + for _, f := range projected.Fields() { + if _, ok := f.Type.(iceberg.PrimitiveType); ok { Review Comment: Could the default set skip geometry and geography columns, and `validateNDVColumns` reject a named one with `ErrInvalidArgument`, plus a geo case in the tests? Both are `PrimitiveType`s, but `getArrowValueAsIcebergLiteral` has no case for the WKB array ([fallthrough](https://github.com/apache/iceberg-go/blob/5ec63084d3e328bf61cab878b99c117592947d7f/table/partitioned_fanout_writer.go#L1349-L1352)). On a V3 table with `id int32, geo geometry`, `ComputeNDVs(ctx, 0)` fails with `unsupported value type: geoarrow.WKBBytes` and returns no sketches. Java's default [keeps geo columns](https://github.com/apache/iceberg/blob/7f07cfb6ec4444706af66e8f9c7c5e83c1ff5071/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ComputeTableStatsSparkAction.java#L132-L142), so sketching the WKB bytes is the other option. Either way, one geo column shouldn't fail the whole call. ########## table/stats_ndv.go: ########## @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "slices" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table/stats" +) + +// NDVResult is the number-of-distinct-values sketch computed for one column. +type NDVResult struct { + // FieldID is the Iceberg field ID of the column. + FieldID int + // NDV is the estimated number of distinct non-null values. + NDV int64 + // Sketch is the serialized DataSketches Theta sketch, i.e. the body of an + // apache-datasketches-theta-v1 statistics blob. + Sketch []byte +} + +// ComputeNDVs scans a snapshot and computes a DataSketches Theta sketch per +// top-level primitive column, returning the estimated number of distinct values +// keyed by field ID. +// +// When snapshotID is 0 the current snapshot is used. When columns is empty every +// top-level primitive column is analyzed, matching Iceberg-Java's +// ComputeTableStats default. Naming an unknown column, a non-primitive column, +// or a column nested inside a struct is an error. +// +// Values are fed to the sketch using Iceberg's single-value serialization (the +// same encoding used for manifest bounds), so estimates are comparable with +// sketches produced by other Iceberg implementations. Null values are not +// counted. +func (t Table) ComputeNDVs(ctx context.Context, snapshotID int64, columns ...string) (map[int]NDVResult, error) { + opts := []ScanOption{WithSnapshotID(snapshotID)} + if len(columns) > 0 { + opts = append(opts, WithSelectedFields(columns...)) + } + scan := t.Scan(opts...) + + // Resolve the snapshot so an empty or missing target is a clear error rather + // than a silently empty result. + if _, err := scan.ResolveSnapshot(); err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + + projected, err := scan.Projection() + if err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + + if err := validateNDVColumns(projected, columns); err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + + // Restrict to top-level primitive fields; nested fields are not sketched, + // matching the reference implementation. + primitives := make(map[string]iceberg.NestedField) + for _, f := range projected.Fields() { + if _, ok := f.Type.(iceberg.PrimitiveType); ok { + primitives[f.Name] = f + } + } + if len(primitives) == 0 { + return map[int]NDVResult{}, nil + } + + arrowSchema, records, err := scan.ToArrowRecords(ctx) + if err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + + // Map each arrow column index to the target field and its sketch, matching + // by name against the projected schema (the returned arrow schema carries no + // field-id metadata). + targets := make(map[int]*ndvTarget) + for i, f := range arrowSchema.Fields() { + nf, ok := primitives[f.Name] + if !ok { + continue + } + targets[i] = &ndvTarget{field: nf, sketch: stats.NewSketch()} + } + + for rec, err := range records { + if err != nil { + return nil, fmt.Errorf("compute NDVs: %w", err) + } + if err := addRecordToSketches(rec, targets); err != nil { + rec.Release() + + return nil, fmt.Errorf("compute NDVs: %w", err) + } + rec.Release() + } + + results := make(map[int]NDVResult, len(targets)) + for _, tgt := range targets { + results[tgt.field.ID] = NDVResult{ + FieldID: tgt.field.ID, + NDV: tgt.sketch.NDV(), + Sketch: tgt.sketch.Serialize(), + } + } + + return results, nil +} + +// validateNDVColumns rejects explicitly named columns that cannot be sketched, +// as Iceberg-Java's ComputeTableStats does, rather than silently dropping them. +func validateNDVColumns(projected *iceberg.Schema, columns []string) error { + for _, name := range columns { + f, ok := projected.FindFieldByName(name) + if !ok { Review Comment: This branch is reachable. `ComputeNDVs(ctx, 0, "*")` lands here, because [`WithSelectedFields`](https://github.com/apache/iceberg-go/blob/5ec63084d3e328bf61cab878b99c117592947d7f/table/table.go#L1390-L1398) treats `"*"` as all columns and skips the select, so we should keep it. A plain unknown name does fail earlier, in [`Schema.Select`](https://github.com/apache/iceberg-go/blob/5ec63084d3e328bf61cab878b99c117592947d7f/schema.go#L649-L653), with `ErrInvalidSchema` rather than `ErrInvalidArgument`. That's the sentinel `TestComputeNDVsUnknownColumnErrors` could assert. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
