laskoviymishka commented on code in PR #1501: URL: https://github.com/apache/iceberg-go/pull/1501#discussion_r3658635203
########## table/scan_metrics.go: ########## @@ -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 table + +import ( + "encoding/json" + "slices" + "strings" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +// scanMetricsAccumulator gathers scan-planning counts. Every field is written +// from a single goroutine — the manifest counts in +// fetchPartitionSpecFilteredManifestsWithSchema and the result/delete counts in +// planFilesLocal after collectManifestEntriesWithSchema's concurrency barrier — +// so plain integers are race-free; no field is touched from the concurrent +// manifest workers. +// +// The scanned/skipped manifest counts measure partition-spec pruning (the +// filter applied while listing manifests). The per-entry skipped-data-files / +// skipped-delete-files counts (which would be incremented inside the concurrent +// openManifest loop, and would use atomics) and indexed-delete-files are left +// for a follow-up and omitted rather than reported as zero. +type scanMetricsAccumulator struct { + totalDataManifests int64 + totalDeleteManifests int64 + scannedDataManifests int64 + scannedDeleteManifests int64 + skippedDataManifests int64 + skippedDeleteManifests int64 + + resultDataFiles int64 + resultDeleteFiles int64 + totalFileSize int64 + totalDeleteFileSize int64 + positionalDeleteFiles int64 + equalityDeleteFiles int64 + dvs int64 +} + +func counterCount(v int64) *metrics.CounterResult { + return metrics.NewCounterResult(metrics.UnitCount, v) +} + +func counterBytes(v int64) *metrics.CounterResult { + return metrics.NewCounterResult(metrics.UnitBytes, v) +} + +// applyResultDeleteMetrics derives the result-scoped delete-file metrics from +// the planned tasks. Following Java's ScanMetricsUtil.fileTask, every delete +// file is counted once per data-file task it applies to — a delete covering N +// data files is counted N times — with no cross-task dedup, so these +// Java-parity field names stay directly comparable with what Java consumers +// report. Sizes accumulate the same way. +// +// A deletion vector's size is its per-blob content_size_in_bytes (Java's +// ScanTaskUtil.contentSizeInBytes), not the size of the Puffin file that holds +// it: DVWriter.Flush packs one DV blob per data file into a single Puffin file, +// so the Puffin length would over-count. +func (acc *scanMetricsAccumulator) applyResultDeleteMetrics(tasks []FileScanTask) { + for _, t := range tasks { + for _, df := range t.DeleteFiles { + acc.positionalDeleteFiles++ + acc.totalDeleteFileSize += df.FileSizeBytes() + } + for _, df := range t.EqualityDeleteFiles { + acc.equalityDeleteFiles++ + acc.totalDeleteFileSize += df.FileSizeBytes() + } + for _, df := range t.DeletionVectorFiles { + acc.dvs++ + if csb := df.ContentSizeInBytes(); csb != nil { + acc.totalDeleteFileSize += *csb + } + } + } + + acc.resultDeleteFiles = acc.positionalDeleteFiles + acc.equalityDeleteFiles + acc.dvs +} + +// buildScanReport assembles the ScanReport for a completed planning operation. +// schema is the schema the scan planned against, threaded in from the planner so +// the report describes exactly what was used rather than re-resolving it here. +// Only the metrics that are actually measured are populated; unmeasured ones +// (e.g. skipped-data-files, indexed-delete-files) are left unset and omitted. +func (scan *Scan) buildScanReport(acc *scanMetricsAccumulator, schema *iceberg.Schema, planning time.Duration) metrics.ScanReport { + ids, names := scan.projectedFields(schema) + + var snapshotID int64 + if snap := scan.Snapshot(); snap != nil { + snapshotID = snap.SnapshotID + } + + // Serialize the scan's row filter as Expression JSON, first sanitizing it so + // predicate literals (which can be user data) never reach a reporter or REST + // metrics sink. On any error we leave Filter unset so ScanReport.MarshalJSON + // falls back to always-true rather than failing the scan or, worse, emitting + // unsanitized literals. + var filter json.RawMessage + if scan.rowFilter != nil { + if sanitized, err := iceberg.SanitizeExpression(scan.rowFilter); err == nil { + if raw, err := json.Marshal(sanitized); err == nil { + filter = raw + } + } + } + + return metrics.ScanReport{ + TableName: strings.Join(scan.identifier, "."), + SnapshotID: snapshotID, + SchemaID: schema.ID, + ProjectedFieldIDs: ids, + ProjectedFieldNames: names, + Filter: filter, + Metrics: metrics.ScanMetricsResult{ + TotalPlanningDuration: metrics.NewNanosTimerResult(1, planning.Nanoseconds()), + ResultDataFiles: counterCount(acc.resultDataFiles), + ResultDeleteFiles: counterCount(acc.resultDeleteFiles), + TotalDataManifests: counterCount(acc.totalDataManifests), + TotalDeleteManifests: counterCount(acc.totalDeleteManifests), + ScannedDataManifests: counterCount(acc.scannedDataManifests), + ScannedDeleteManifests: counterCount(acc.scannedDeleteManifests), + SkippedDataManifests: counterCount(acc.skippedDataManifests), + SkippedDeleteManifests: counterCount(acc.skippedDeleteManifests), + TotalFileSizeInBytes: counterBytes(acc.totalFileSize), + TotalDeleteFileSizeInBytes: counterBytes(acc.totalDeleteFileSize), + EqualityDeleteFiles: counterCount(acc.equalityDeleteFiles), + PositionalDeleteFiles: counterCount(acc.positionalDeleteFiles), + DVs: counterCount(acc.dvs), + }, + } +} + +// projectedFields resolves the scan's projection to the field ids and dotted +// column names it reads, matching Java's TypeUtil.getProjectedIds + +// Schema.findColumnName: nested fields are reported by id and full dotted path, +// not just top-level columns. schema is the schema the scan planned against and +// is used to resolve names; ids are returned sorted for deterministic output. +// Ids not resolvable to a name in schema (e.g. reserved row-lineage columns) are +// skipped so ids and names stay aligned. +func (scan *Scan) projectedFields(schema *iceberg.Schema) ([]int, []string) { + projected, err := scan.Projection() Review Comment: threading the schema through `buildScanReport` is exactly what I wanted, thanks — but `projectedFields` still re-derives the ids from `scan.Projection()` and only uses the schema param for the names. that leaves a seam: if `Projection()` fails for a reason planning didn't hit (the row-lineage version check, say), we silently return nil, nil and emit a report with no projected fields even though planning succeeded. since we already have the exact schema the scan planned against, I'd walk that directly (`IndexByID(schema)` / iterate its fields) and drop the `scan.Projection()` call. wdyt? ########## table/scan_metrics_test.go: ########## @@ -0,0 +1,277 @@ +// 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" + "testing" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func metricsTestMetadata(t *testing.T) Metadata { + t.Helper() + schema := iceberg.NewSchema(7, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String}, + ) + meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, + "mem://bucket/tbl", iceberg.Properties{}) + require.NoError(t, err) + + return meta +} + +func TestBuildScanReport(t *testing.T) { + meta := metricsTestMetadata(t) + scan := &Scan{ + metadata: meta, + identifier: Identifier{"db", "tbl"}, + selectedFields: []string{"*"}, + caseSensitive: true, + } + + acc := &scanMetricsAccumulator{ + totalDataManifests: 3, + scannedDataManifests: 2, + skippedDataManifests: 1, + totalDeleteManifests: 1, + resultDataFiles: 10, + totalFileSize: 2048, + positionalDeleteFiles: 1, + equalityDeleteFiles: 2, + dvs: 1, + resultDeleteFiles: 4, + totalDeleteFileSize: 512, + } + + sr := scan.buildScanReport(acc, meta.CurrentSchema(), 5*time.Millisecond) + + assert.Equal(t, "db.tbl", sr.TableName) + assert.Equal(t, meta.CurrentSchema().ID, sr.SchemaID) + assert.Equal(t, []int{1, 2}, sr.ProjectedFieldIDs) + assert.Equal(t, []string{"id", "data"}, sr.ProjectedFieldNames) + // No row filter set: Filter stays unset and defaults to always-true on the wire. + assert.Nil(t, sr.Filter) + + m := sr.Metrics + require.NotNil(t, m.TotalPlanningDuration) + assert.Equal(t, metrics.TimeUnitNanoseconds, m.TotalPlanningDuration.TimeUnit) + assert.Equal(t, (5 * time.Millisecond).Nanoseconds(), m.TotalPlanningDuration.TotalDuration) + assert.Equal(t, int64(1), m.TotalPlanningDuration.Count) + + require.NotNil(t, m.TotalDataManifests) + assert.Equal(t, metrics.UnitCount, m.TotalDataManifests.Unit) + assert.Equal(t, int64(3), m.TotalDataManifests.Value) + assert.Equal(t, int64(2), m.ScannedDataManifests.Value) + assert.Equal(t, int64(1), m.SkippedDataManifests.Value) + assert.Equal(t, int64(10), m.ResultDataFiles.Value) + assert.Equal(t, int64(4), m.ResultDeleteFiles.Value) + assert.Equal(t, int64(1), m.PositionalDeleteFiles.Value) + assert.Equal(t, int64(2), m.EqualityDeleteFiles.Value) + assert.Equal(t, int64(1), m.DVs.Value) + + require.NotNil(t, m.TotalFileSizeInBytes) + assert.Equal(t, metrics.UnitBytes, m.TotalFileSizeInBytes.Unit) + assert.Equal(t, int64(2048), m.TotalFileSizeInBytes.Value) + assert.Equal(t, int64(512), m.TotalDeleteFileSizeInBytes.Value) + + // Unmeasured metrics are left unset (omitted on the wire). + assert.Nil(t, m.SkippedDataFiles) + assert.Nil(t, m.IndexedDeleteFiles) +} + +func TestProjectedFieldsSelectedSubset(t *testing.T) { + meta := metricsTestMetadata(t) + scan := &Scan{metadata: meta, selectedFields: []string{"data"}, caseSensitive: true} + + ids, names := scan.projectedFields(meta.CurrentSchema()) + assert.Equal(t, []int{2}, ids) + assert.Equal(t, []string{"data"}, names) +} + +func TestProjectedFieldsNested(t *testing.T) { + // Selecting a struct column must report its nested fields by id and full + // dotted path, not just the top-level column (Java's TypeUtil.getProjectedIds + // + Schema.findColumnName). + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "location", Required: false, Type: &iceberg.StructType{ + FieldList: []iceberg.NestedField{ + {ID: 3, Name: "lat", Type: iceberg.PrimitiveTypes.Float64, Required: false}, + {ID: 4, Name: "lon", Type: iceberg.PrimitiveTypes.Float64, Required: false}, + }, + }}, + ) + meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, + "mem://bucket/tbl", iceberg.Properties{}) + require.NoError(t, err) + + scan := &Scan{metadata: meta, selectedFields: []string{"location"}, caseSensitive: true} + + ids, names := scan.projectedFields(meta.CurrentSchema()) + assert.Equal(t, []int{2, 3, 4}, ids) + assert.Equal(t, []string{"location", "location.lat", "location.lon"}, names) +} + +func TestApplyResultDeleteMetrics(t *testing.T) { + posA := &mockDataFile{path: "s3://b/pos-a.parquet", filesize: 100} + posB := &mockDataFile{path: "s3://b/pos-b.parquet", filesize: 200} + eq := &mockDataFile{path: "s3://b/eq.parquet", filesize: 40} + // A DV is a Puffin file; its size is the per-blob content_size_in_bytes, not + // the whole Puffin file size, so filesize (the Puffin length) must be ignored. + dv := &dvMockDataFile{ + mockDataFile: mockDataFile{path: "s3://b/deletes.puffin", filesize: 9999}, + referencedDataFile: strPtr("s3://b/data-2.parquet"), + contentSizeInBytes: int64Ptr(10), + } + + // posA applies to both data-file tasks: matching Java's per-task + // accumulation, it must be counted (and sized) once per task, i.e. twice. + tasks := []FileScanTask{ + { + File: &mockDataFile{path: "s3://b/data-1.parquet", filesize: 1000}, + DeleteFiles: []iceberg.DataFile{posA, posB}, + }, + { + File: &mockDataFile{path: "s3://b/data-2.parquet", filesize: 2000}, + DeleteFiles: []iceberg.DataFile{posA}, + EqualityDeleteFiles: []iceberg.DataFile{eq}, + DeletionVectorFiles: []iceberg.DataFile{dv}, + }, + } + + var acc scanMetricsAccumulator + acc.applyResultDeleteMetrics(tasks) + + assert.Equal(t, int64(3), acc.positionalDeleteFiles, "posA counted once per task it applies to") + assert.Equal(t, int64(1), acc.equalityDeleteFiles) + assert.Equal(t, int64(1), acc.dvs) + assert.Equal(t, int64(5), acc.resultDeleteFiles) + // task 1: 100 (posA) + 200 (posB); task 2: 100 (posA) + 40 (eq) + 10 (dv + // content size, not 9999) = 450 total. + assert.Equal(t, int64(450), acc.totalDeleteFileSize) +} + +func TestApplyResultDeleteMetricsDVsShareOnePuffin(t *testing.T) { + // DVWriter.Flush writes one Puffin file with a separate DV blob per data + // file, so both DVs below share a FilePath() and differ only by the data + // file they reference. Each is planned on its own data-file task, so per-task + // counting reports 2 DVs, and each blob's own content size must be summed + // (never the shared Puffin file length). + puffin := "s3://b/deletes.puffin" + ref1, ref2 := "s3://b/data-1.parquet", "s3://b/data-2.parquet" + dv1 := &dvMockDataFile{ + mockDataFile: mockDataFile{path: puffin, filesize: 9999}, + referencedDataFile: &ref1, + contentSizeInBytes: int64Ptr(30), + } + dv2 := &dvMockDataFile{ + mockDataFile: mockDataFile{path: puffin, filesize: 9999}, + referencedDataFile: &ref2, + contentSizeInBytes: int64Ptr(70), + } + + tasks := []FileScanTask{ + {File: &mockDataFile{path: ref1, filesize: 1000}, DeletionVectorFiles: []iceberg.DataFile{dv1}}, + {File: &mockDataFile{path: ref2, filesize: 2000}, DeletionVectorFiles: []iceberg.DataFile{dv2}}, + } + + var acc scanMetricsAccumulator + acc.applyResultDeleteMetrics(tasks) + + assert.Equal(t, int64(2), acc.dvs, "two DVs in one Puffin file must not collapse") + assert.Equal(t, int64(2), acc.resultDeleteFiles) + assert.Equal(t, int64(100), acc.totalDeleteFileSize, "sum of per-blob content sizes, not Puffin file size") +} + +func TestBuildScanReportSetsFilter(t *testing.T) { + meta := metricsTestMetadata(t) + scan := &Scan{ + metadata: meta, + identifier: Identifier{"db", "tbl"}, + selectedFields: []string{"*"}, + caseSensitive: true, + rowFilter: iceberg.AlwaysFalse{}, + } + + sr := scan.buildScanReport(&scanMetricsAccumulator{}, meta.CurrentSchema(), time.Millisecond) + assert.JSONEq(t, `false`, string(sr.Filter), "Filter must reflect the real row filter, not default to always-true") +} + +func TestBuildScanReportSanitizesFilter(t *testing.T) { + // Predicate literals can be user data, so the emitted filter must be + // sanitized (Java's ExpressionUtil.sanitize) before it reaches a reporter. + meta := metricsTestMetadata(t) + scan := &Scan{ + metadata: meta, + identifier: Identifier{"db", "tbl"}, + selectedFields: []string{"*"}, + caseSensitive: true, + rowFilter: iceberg.EqualTo(iceberg.Reference("data"), "secret-value"), + } + + sr := scan.buildScanReport(&scanMetricsAccumulator{}, meta.CurrentSchema(), time.Millisecond) + require.NotNil(t, sr.Filter) + assert.NotContains(t, string(sr.Filter), "secret-value", "literal must not leak into the report") + assert.Contains(t, string(sr.Filter), "data", "column reference is preserved") + assert.Contains(t, string(sr.Filter), "(redacted)") +} + +func TestPlanFilesNoSnapshotEmitsNoReport(t *testing.T) { Review Comment: the real-snapshot counter test (`TestFetchManifestCountersWithRealSnapshot`) is in and covers the manifest accounting nicely — the one path still untested is the emission happy path. nothing runs `PlanFiles` against a real snapshot with a live reporter and asserts the emitted report carries the right `SnapshotID` and non-zero counters. `buildScanReport` is only unit-tested with a no-snapshot scan, so the `snapshotID = snap.SnapshotID` branch never actually runs, and a regression in the `Snapshot() != nil` gate or the id population would be invisible. could we add a positive sibling to this test that plans through a real snapshot with an `InMemoryReporter` and asserts `Reports()` has one report with the expected id? ########## visitors.go: ########## @@ -520,3 +520,84 @@ func (c columnNameTranslator) VisitBound(pred BoundPredicate) BooleanExpression panic(fmt.Errorf("%w: unsupported predicate: %s", ErrNotImplemented, pred)) } } + +// sanitizedLiteralMask is the placeholder substituted for every literal value in +// a sanitized expression. It carries no information about the original value. +const sanitizedLiteralMask = "(redacted)" + +// SanitizeExpression returns a copy of expr with every predicate literal replaced +// by an opaque placeholder while preserving the boolean structure, column +// references, and operations. It lets a filter be emitted somewhere untrusted +// (e.g. a metrics ScanReport shipped to a REST sink) without leaking the literal +// values a user scanned with, mirroring the intent of Java's ExpressionUtil.sanitize. +// +// Unlike Java, which substitutes type-shaped placeholders (e.g. "(2-digit-int)"), +// every literal is masked with the same marker; the goal here is to withhold the +// values, not to preserve their shape. Set predicates (IN / NOT IN) keep their +// arity so the operation is not misrepresented, but the members are masked. +func SanitizeExpression(expr BooleanExpression) (BooleanExpression, error) { Review Comment: not blocking — sanitizing before the filter ships is great, and further than I asked for in round 1. now that `SanitizeExpression` is exported, the `(redacted)` marker is effectively a public contract, and it diverges from Java's type-shaped placeholders, so a catalog can't correlate these filters across clients. the doc comment already flags it, which is the right call — I'd just have the godoc say the guarantee is "no user value leaks," not structural parity with Java, so we're free to change the format later without it being a breaking change. a `NOT IN` case in the test wouldn't hurt either, since it rides the same set-predicate branch as `IN`. -- 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]
