laskoviymishka commented on code in PR #1501: URL: https://github.com/apache/iceberg-go/pull/1501#discussion_r3629456371
########## table/scan_metrics.go: ########## @@ -0,0 +1,172 @@ +// 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 ( + "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. Each delete file is counted once by path across all the +// data files it applies to, split by kind (positional, equality, deletion +// vector); positional deletes suppressed by a deletion vector never appear on a +// task, so they are excluded. This keeps the counts consistent with +// result-data-files and total-file-size-in-bytes. +func (acc *scanMetricsAccumulator) applyResultDeleteMetrics(tasks []FileScanTask) { + posDeletes := make(map[string]int64) + eqDeletes := make(map[string]int64) + dvs := make(map[string]int64) + for _, t := range tasks { + for _, df := range t.DeleteFiles { + posDeletes[df.FilePath()] = df.FileSizeBytes() + } + for _, df := range t.EqualityDeleteFiles { + eqDeletes[df.FilePath()] = df.FileSizeBytes() + } + for _, df := range t.DeletionVectorFiles { + dvs[df.FilePath()] = df.FileSizeBytes() + } + } + + acc.positionalDeleteFiles = int64(len(posDeletes)) + acc.equalityDeleteFiles = int64(len(eqDeletes)) + acc.dvs = int64(len(dvs)) + acc.resultDeleteFiles = acc.positionalDeleteFiles + acc.equalityDeleteFiles + acc.dvs + for _, deletes := range []map[string]int64{posDeletes, eqDeletes, dvs} { + for _, size := range deletes { + acc.totalDeleteFileSize += size + } + } +} + +// buildScanReport assembles the ScanReport for a completed planning operation. +// 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, planning time.Duration) metrics.ScanReport { + // Resolve the schema the scan actually planned against (the snapshot schema + // for snapshot/as-of scans, the current schema otherwise). Best-effort: a + // report must never fail the scan, so fall back to the current schema. + schema, err := scan.effectiveSchema() + if err != nil || schema == nil { + schema = scan.metadata.CurrentSchema() + } + + ids, names := scan.projectedFields(schema) + + var snapshotID int64 + if snap := scan.Snapshot(); snap != nil { + snapshotID = snap.SnapshotID + } + + return metrics.ScanReport{ + TableName: strings.Join(scan.identifier, "."), Review Comment: `buildScanReport` never sets `Filter`, so the `MarshalJSON` fallback emits `"filter": true` unconditionally — a scan with a real `rowFilter` still reports `filter: true`, which is misleading once these land in a REST catalog's metrics table. The infra is already there: every concrete `BooleanExpression` has `MarshalJSON` (expr_json.go). I'd set `Filter` from `scan.rowFilter` here, falling back to always-true on marshal error. Java sanitizes literals via `ExpressionUtil` first and we don't have an equivalent yet, so I'm fine leaving sanitization as a follow-up — but the field needs to reflect the actual filter. ########## table/scanner.go: ########## @@ -653,10 +654,13 @@ func (scan *Scan) fetchPartitionSpecFilteredManifests(ctx context.Context) ([]ic return nil, err } Review Comment: This wrapper allocates a fresh `&scanMetricsAccumulator{}` and throws it away — the manifest counts it records go nowhere. Fine today since this path has no reporter behind it, but it's a quiet trap: a future caller wiring this path to a report would silently get zeroed manifest counters with no hint why. I'd either drop the accumulator param from this path, or leave a comment making it explicit that the counts are intentionally discarded here. ########## table/scan_metrics.go: ########## @@ -0,0 +1,172 @@ +// 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 ( + "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. Each delete file is counted once by path across all the +// data files it applies to, split by kind (positional, equality, deletion +// vector); positional deletes suppressed by a deletion vector never appear on a +// task, so they are excluded. This keeps the counts consistent with +// result-data-files and total-file-size-in-bytes. +func (acc *scanMetricsAccumulator) applyResultDeleteMetrics(tasks []FileScanTask) { + posDeletes := make(map[string]int64) + eqDeletes := make(map[string]int64) + dvs := make(map[string]int64) + for _, t := range tasks { + for _, df := range t.DeleteFiles { + posDeletes[df.FilePath()] = df.FileSizeBytes() + } + for _, df := range t.EqualityDeleteFiles { + eqDeletes[df.FilePath()] = df.FileSizeBytes() + } + for _, df := range t.DeletionVectorFiles { + dvs[df.FilePath()] = df.FileSizeBytes() Review Comment: I think there's a real undercount here for deletion vectors. `DVWriter.Flush` writes one Puffin file per flush with a DV blob per data file, so every DV `DataFile` from a flush shares the same `FilePath()` (the puffin location) and differs only by `ReferencedDataFile()`. Keying `dvs` by `FilePath()` collapses N DVs into 1, and `resultDeleteFiles` comes out off by N-1. Java keys `dvByPath` by `referencedDataFile()`, not location — I'd key the DV map by `df.ReferencedDataFile()`. Same map, second bug: the size is `df.FileSizeBytes()`, which is the whole puffin file, not the per-blob DV length. Java's `ScanTaskUtil.contentSizeInBytes` uses `contentSizeInBytes()` for DVs. `DataFile` already carries `ContentSizeInBytes()` (populated in dv_writer.go), so I'd use that (nil-guarded) for the DV size. While we're here — worth renaming the local `dvs` map, since it shadows the `acc.dvs` field and the other two are `posDeletes`/`eqDeletes` for exactly that reason. And `TestApplyResultDeleteMetrics` should grow a case with two DVs sharing one puffin path so this stays covered once fixed. ########## table/scanner.go: ########## @@ -812,6 +835,30 @@ func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) { return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, scan.planningMode) } + start := time.Now() Review Comment: `start` is captured here, but `time.Since(start)` is computed inside the `Report` call below, so `TotalPlanningDuration` also includes `buildScanReport`'s schema re-resolution and projection. I'd snap the elapsed time right after `planFilesLocal` returns and pass that in, so the duration reflects just planning. ########## table/scanner.go: ########## @@ -812,6 +835,30 @@ func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) { return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, scan.planningMode) } + start := time.Now() + var acc scanMetricsAccumulator + + results, err := scan.planFilesLocal(ctx, &acc) + if err != nil { + return nil, err + } + + // Building the report is skipped for the no-op reporter — the opt-in default + // set by Table.Scan, and the nil-reporter case for scans constructed directly + // in tests (see Reporter, which maps nil to NopReporter). A NopReporter + // discards the report, so assembling one would be pure overhead. + rep := scan.Reporter() + if _, isNop := rep.(metrics.NopReporter); !isNop { + rep.Report(ctx, scan.buildScanReport(&acc, time.Since(start))) + } + + return results, nil +} + +// planFilesLocal performs local scan planning: it reads and filters the +// snapshot's manifests, builds the matching FileScanTasks, and records planning +// metrics into acc for the caller to report. +func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulator) ([]FileScanTask, error) { scan.planIO = nil Review Comment: Two small things on `planFilesLocal`: it mutates `scan.planIO = nil` as a side effect that the doc comment doesn't mention, and it returns `(nil, nil)` rather than `([]FileScanTask{}, nil)` on empty manifests. Neither is wrong, but I'd note the `planIO` reset in the comment and pick nil-vs-empty deliberately so callers know which to expect. ########## table/scanner.go: ########## @@ -812,6 +835,30 @@ func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) { return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, scan.planningMode) } + start := time.Now() + var acc scanMetricsAccumulator + + results, err := scan.planFilesLocal(ctx, &acc) + if err != nil { + return nil, err + } + + // Building the report is skipped for the no-op reporter — the opt-in default + // set by Table.Scan, and the nil-reporter case for scans constructed directly + // in tests (see Reporter, which maps nil to NopReporter). A NopReporter + // discards the report, so assembling one would be pure overhead. + rep := scan.Reporter() + if _, isNop := rep.(metrics.NopReporter); !isNop { Review Comment: This guard only catches a bare `metrics.NopReporter` — `Combine(NopReporter{})` returns a `*compositeReporter` and slips right through, so the comment above about avoiding "pure overhead" isn't quite true for a wrapped nop. Minor, but I'd either add a small `isNop`-style interface the composite can answer, or soften the comment so it doesn't over-promise. ########## table/scan_metrics.go: ########## @@ -0,0 +1,172 @@ +// 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 ( + "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. Each delete file is counted once by path across all the +// data files it applies to, split by kind (positional, equality, deletion +// vector); positional deletes suppressed by a deletion vector never appear on a +// task, so they are excluded. This keeps the counts consistent with +// result-data-files and total-file-size-in-bytes. +func (acc *scanMetricsAccumulator) applyResultDeleteMetrics(tasks []FileScanTask) { + posDeletes := make(map[string]int64) + eqDeletes := make(map[string]int64) + dvs := make(map[string]int64) + for _, t := range tasks { + for _, df := range t.DeleteFiles { + posDeletes[df.FilePath()] = df.FileSizeBytes() + } + for _, df := range t.EqualityDeleteFiles { + eqDeletes[df.FilePath()] = df.FileSizeBytes() + } + for _, df := range t.DeletionVectorFiles { + dvs[df.FilePath()] = df.FileSizeBytes() + } + } + + acc.positionalDeleteFiles = int64(len(posDeletes)) + acc.equalityDeleteFiles = int64(len(eqDeletes)) + acc.dvs = int64(len(dvs)) + acc.resultDeleteFiles = acc.positionalDeleteFiles + acc.equalityDeleteFiles + acc.dvs + for _, deletes := range []map[string]int64{posDeletes, eqDeletes, dvs} { + for _, size := range deletes { + acc.totalDeleteFileSize += size + } + } +} + +// buildScanReport assembles the ScanReport for a completed planning operation. +// 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, planning time.Duration) metrics.ScanReport { + // Resolve the schema the scan actually planned against (the snapshot schema + // for snapshot/as-of scans, the current schema otherwise). Best-effort: a + // report must never fail the scan, so fall back to the current schema. + schema, err := scan.effectiveSchema() Review Comment: `buildScanReport` re-resolves the schema with `effectiveSchema()` and then silently falls back to `CurrentSchema()` on error — but `planFilesLocal` already resolved the schema it actually planned against. If the two disagree (or the fallback fires), the report describes a different schema than the one used for planning. I'd thread the planning schema through into `buildScanReport` instead of re-resolving. It also drops a redundant `Schemas()` scan per `PlanFiles`. ########## table/scanner.go: ########## @@ -812,6 +835,30 @@ func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) { return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, scan.planningMode) } + start := time.Now() + var acc scanMetricsAccumulator + + results, err := scan.planFilesLocal(ctx, &acc) + if err != nil { + return nil, err + } + + // Building the report is skipped for the no-op reporter — the opt-in default + // set by Table.Scan, and the nil-reporter case for scans constructed directly + // in tests (see Reporter, which maps nil to NopReporter). A NopReporter + // discards the report, so assembling one would be pure overhead. + rep := scan.Reporter() + if _, isNop := rep.(metrics.NopReporter); !isNop { + rep.Report(ctx, scan.buildScanReport(&acc, time.Since(start))) Review Comment: `planFilesLocal` returns `(nil, nil)` when there's no snapshot or every manifest is pruned, but we still emit a report here with `SnapshotID: 0` — a zero-value int64, not a real snapshot id. Java doesn't emit a scan report in the no-snapshot case at all. I'd guard on `scan.Snapshot() == nil` before emitting so we don't push bogus zero-snapshot records into the reporter. wdyt? ########## table/scan_metrics_test.go: ########## @@ -0,0 +1,187 @@ +// 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, 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) + + 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 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} + dv := &mockDataFile{path: "s3://b/dv.puffin", filesize: 10} + + // posA applies to both data files: it must be counted (and sized) once. + 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) + Review Comment: This test scans empty metadata, so every manifest counter is zero — which means the scanned/skipped increment logic in `fetchPartitionSpecFilteredManifestsWithSchema` has no real coverage. A swapped scanned/skipped or data/delete branch would sail through green. Could we add a case with a real snapshot that has manifests and assert `scanned + skipped == total` (and that delete manifests land in the delete counters)? That's the part most likely to regress silently. ########## table/scan_metrics.go: ########## @@ -0,0 +1,172 @@ +// 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 ( + "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. Each delete file is counted once by path across all the +// data files it applies to, split by kind (positional, equality, deletion +// vector); positional deletes suppressed by a deletion vector never appear on a +// task, so they are excluded. This keeps the counts consistent with +// result-data-files and total-file-size-in-bytes. +func (acc *scanMetricsAccumulator) applyResultDeleteMetrics(tasks []FileScanTask) { Review Comment: One parity note: Java's `ScanMetricsUtil.fileTask` counts `resultDeleteFiles` per data-file task with no cross-task dedup, so a positional delete covering 5 data files counts 5 there and 1 here. The count-once-by-path choice is defensible as a result-scoped metric, but since it diverges from Java I'd call it out in this comment so the next person doesn't "fix" it into a mismatch. -- 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]
