You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/19 18:32:05 UTC
(iceberg-go) branch main updated: feat(table): add files metadata tables (#1744)
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 6d34de587 feat(table): add files metadata tables (#1744)
6d34de587 is described below
commit 6d34de58762eb5a005715d635fa018ac20e654d3
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Wed Aug 19 20:31:59 2026 +0200
feat(table): add files metadata tables (#1744)
* feat(table): add data files metadata table
* chore: fix lint findings
* fix: support evolved partition specs in data files
* chore: fix lint findings
* fix(table): support decimal partition metadata values
* perf(table): stream metadata file inspection batches
* fix(table): validate evolved partition fields
* fix(table): satisfy lint for partition metadata tests
* fix(table): harden data files metadata inspection
* fix(table): remove deprecated distinct counts
* fix(table): validate inspect partition transforms
* feat(table): add files metadata tables
Signed-off-by: Minh Vu <vu...@gmail.com>
* refactor(table): share files inspection path
* test(table): cover deleted files in metadata tables
* fix(table): preserve yielded metadata batch ownership
* refactor(table): streamline files metadata inspection
* refactor(table): simplify metadata manifest iteration
* ci: assert metadata reader ownership
* ci: share ownership assertion command
---------
Signed-off-by: Minh Vu <vu...@gmail.com>
---
.github/workflows/go-ci.yml | 3 +
Makefile | 7 +-
table/inspect_delete_files.go | 21 +---
table/inspect_files.go | 179 +++++++++++++++++++++++---
table/inspect_internal_test.go | 280 ++++++++++++++++++++++++++++++++++++++++-
5 files changed, 447 insertions(+), 43 deletions(-)
diff --git a/.github/workflows/go-ci.yml b/.github/workflows/go-ci.yml
index 261ae81c0..962c27043 100644
--- a/.github/workflows/go-ci.yml
+++ b/.github/workflows/go-ci.yml
@@ -93,6 +93,9 @@ jobs:
# Race detector is opt-in per package/test.
- name: Run race detector
run: go test -race -v ./codec/... ./table/...
+ - name: Run Arrow ownership assertions
+ if: matrix.os == 'ubuntu-latest' && matrix.go == '1.26.1'
+ run: make test-assert
- name: Save Go module and build cache
if: github.ref == 'refs/heads/main'
uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
diff --git a/Makefile b/Makefile
index 7ccdfb2c6..0aeafaf4b 100644
--- a/Makefile
+++ b/Makefile
@@ -17,11 +17,14 @@
# golangci-lint version (keep in sync with CI and README)
GOLANGCI_LINT_VERSION := v2.12.2
-.PHONY: test test-race lint lint-install integration-setup integration-setup-spark4 integration-test integration-scanner integration-io integration-rest integration-spark integration-hadoop integration-down integration-logs docs-gen
+.PHONY: test test-assert test-race lint lint-install integration-setup integration-setup-spark4 integration-test integration-scanner integration-io integration-rest integration-spark integration-hadoop integration-down integration-logs docs-gen
test:
go test -v ./...
+test-assert:
+ go test -tags=assert -v -run='^(TestInspectFilesTablesEarlyRelease|TestInspectDataFilesEmitsEmptyBatchWhenAllEntriesAreDeleted|TestInspectDataFilesEmptyTableEarlyRelease)$$' ./table
+
# Race detector is opt-in per package/test.
test-race:
go test -race -v ./codec/... ./metrics/...
@@ -79,4 +82,4 @@ integration-hadoop:
go test -tags=integration -v -run="^TestHadoopIntegration" ./catalog/hadoop/...
integration-hadoop-clean:
- rm -rf /tmp/iceberg-hadoop-warehouse/*
\ No newline at end of file
+ rm -rf /tmp/iceberg-hadoop-warehouse/*
diff --git a/table/inspect_delete_files.go b/table/inspect_delete_files.go
index 7ec64f6d2..8740c619b 100644
--- a/table/inspect_delete_files.go
+++ b/table/inspect_delete_files.go
@@ -19,7 +19,6 @@ package table
import (
"context"
- "fmt"
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/iceberg-go"
@@ -29,28 +28,10 @@ import (
// deletion-vector files in the current snapshot. It reports file metadata and
// does not open or scan the delete-file contents.
func (i InspectTable) DeleteFiles(ctx context.Context) (array.RecordReader, error) {
- partitionType, err := inspectPartitionType(i.tbl.metadata)
- if err != nil {
- return nil, fmt.Errorf("inspect delete files: %w", err)
- }
- schema := DeleteFilesSchema(partitionType)
- arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
- if err != nil {
- return nil, fmt.Errorf("inspect delete files: build arrow schema: %w", err)
- }
-
- rr, err := i.manifestEntryReader(ctx, arrowSchema, true,
+ return i.inspectFiles(ctx, "delete files", DeleteFilesSchema, false,
func(manifest iceberg.ManifestFile) bool {
return manifest.ManifestContent() == iceberg.ManifestContentDeletes
- },
- func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error {
- return appendContentFileRecord(bldr, partitionType, entry.DataFile())
})
- if err != nil {
- return nil, fmt.Errorf("inspect delete files: %w", err)
- }
-
- return rr, nil
}
// DeleteFilesSchema returns the content-file schema used by the delete_files
diff --git a/table/inspect_files.go b/table/inspect_files.go
index 378e0fe42..fe80b4699 100644
--- a/table/inspect_files.go
+++ b/table/inspect_files.go
@@ -31,32 +31,80 @@ import (
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/arrow/scalar"
"github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
"github.com/google/uuid"
)
// DataFiles returns the live data files in the current snapshot. Deleted
// manifest entries are omitted, matching the data_files metadata table.
func (i InspectTable) DataFiles(ctx context.Context) (array.RecordReader, error) {
+ return i.inspectFiles(ctx, "data files", DataFilesSchema, false,
+ func(manifest iceberg.ManifestFile) bool {
+ return manifest.ManifestContent() == iceberg.ManifestContentData
+ })
+}
+
+// Files returns all live data and delete files in the current snapshot.
+// Deleted manifest entries are omitted.
+func (i InspectTable) Files(ctx context.Context) (array.RecordReader, error) {
+ return i.inspectFiles(ctx, "files", FilesSchema, false, nil)
+}
+
+// AllFiles returns the live data and delete files reachable from every
+// snapshot currently tracked by the table. Shared manifests are scanned once,
+// while duplicate file rows from different manifests are preserved.
+func (i InspectTable) AllFiles(ctx context.Context) (array.RecordReader, error) {
+ return i.inspectFiles(ctx, "all files", AllFilesSchema, true, nil)
+}
+
+// AllDataFiles returns the live data files reachable from every snapshot
+// currently tracked by the table.
+func (i InspectTable) AllDataFiles(ctx context.Context) (array.RecordReader, error) {
+ return i.inspectFiles(ctx, "all data files", AllDataFilesSchema, true,
+ func(manifest iceberg.ManifestFile) bool {
+ return manifest.ManifestContent() == iceberg.ManifestContentData
+ })
+}
+
+// AllDeleteFiles returns the live delete files reachable from every snapshot
+// currently tracked by the table.
+func (i InspectTable) AllDeleteFiles(ctx context.Context) (array.RecordReader, error) {
+ return i.inspectFiles(ctx, "all delete files", AllDeleteFilesSchema, true,
+ func(manifest iceberg.ManifestFile) bool {
+ return manifest.ManifestContent() == iceberg.ManifestContentDeletes
+ })
+}
+
+type inspectFilesSchema func(*iceberg.StructType) *iceberg.Schema
+
+func (i InspectTable) inspectFiles(
+ ctx context.Context,
+ name string,
+ schemaFn inspectFilesSchema,
+ allSnapshots bool,
+ includeManifest func(iceberg.ManifestFile) bool,
+) (array.RecordReader, error) {
partitionType, err := inspectPartitionType(i.tbl.metadata)
if err != nil {
- return nil, fmt.Errorf("inspect data files: %w", err)
+ return nil, fmt.Errorf("inspect %s: %w", name, err)
}
- schema := DataFilesSchema(partitionType)
- arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ arrowSchema, err := SchemaToArrowSchema(schemaFn(partitionType), nil, true, false)
if err != nil {
- return nil, fmt.Errorf("inspect data files: build arrow schema: %w", err)
+ return nil, fmt.Errorf("inspect %s: build arrow schema: %w", name, err)
}
appendFile := newInspectContentFileAppender(partitionType)
- rr, err := i.manifestEntryReader(ctx, arrowSchema, true,
- func(manifest iceberg.ManifestFile) bool {
- return manifest.ManifestContent() == iceberg.ManifestContentData
- },
- func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error {
- return appendFile(bldr, entry.DataFile())
- })
+ appendEntry := func(bldr *array.RecordBuilder, entry iceberg.ManifestEntry) error {
+ return appendFile(bldr, entry.DataFile())
+ }
+ var rr array.RecordReader
+ if allSnapshots {
+ rr, err = i.allManifestEntryReader(ctx, arrowSchema, true, includeManifest, appendEntry)
+ } else {
+ rr, err = i.manifestEntryReader(ctx, arrowSchema, true, includeManifest, appendEntry)
+ }
if err != nil {
- return nil, fmt.Errorf("inspect data files: %w", err)
+ return nil, fmt.Errorf("inspect %s: %w", name, err)
}
return rr, nil
@@ -91,6 +139,78 @@ func (i InspectTable) manifestEntryReader(
return nil, err
}
+ return i.manifestEntryReaderFromManifestSource(
+ ctx, arrowSchema, fs, discardDeleted, includeManifest, appendEntry,
+ func(yield func(iceberg.ManifestFile, error) bool) {
+ for _, manifest := range manifests {
+ if !yield(manifest, nil) {
+ return
+ }
+ }
+ }), nil
+}
+
+// allManifestEntryReader scans each manifest reachable from the table's
+// tracked snapshots exactly once. Manifest paths are immutable and unique, so
+// the path is the stable identity used by Java and PyIceberg's all_* tables.
+func (i InspectTable) allManifestEntryReader(
+ ctx context.Context,
+ arrowSchema *arrow.Schema,
+ discardDeleted bool,
+ includeManifest func(iceberg.ManifestFile) bool,
+ appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error,
+) (array.RecordReader, error) {
+ snapshots := i.tbl.metadata.Snapshots()
+ if len(snapshots) == 0 {
+ return array.ReaderFromIter(arrowSchema, emptyInspectRecordBatch(i.alloc, arrowSchema)), nil
+ }
+ if i.tbl.fsF == nil {
+ return nil, errors.New("table file IO is not configured")
+ }
+
+ fs, err := i.tbl.fsF(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ return i.manifestEntryReaderFromManifestSource(
+ ctx, arrowSchema, fs, discardDeleted, includeManifest, appendEntry,
+ func(yield func(iceberg.ManifestFile, error) bool) {
+ seen := make(map[string]struct{})
+ for _, snapshot := range snapshots {
+ if err := ctx.Err(); err != nil {
+ yield(nil, err)
+
+ return
+ }
+ snapshotManifests, err := snapshot.Manifests(fs)
+ if err != nil {
+ yield(nil, fmt.Errorf("read snapshot %d manifests: %w", snapshot.SnapshotID, err))
+
+ return
+ }
+ for _, manifest := range snapshotManifests {
+ if _, ok := seen[manifest.FilePath()]; ok {
+ continue
+ }
+ seen[manifest.FilePath()] = struct{}{}
+ if !yield(manifest, nil) {
+ return
+ }
+ }
+ }
+ }), nil
+}
+
+func (i InspectTable) manifestEntryReaderFromManifestSource(
+ ctx context.Context,
+ arrowSchema *arrow.Schema,
+ fs iceio.IO,
+ discardDeleted bool,
+ includeManifest func(iceberg.ManifestFile) bool,
+ appendEntry func(*array.RecordBuilder, iceberg.ManifestEntry) error,
+ source iter.Seq2[iceberg.ManifestFile, error],
+) array.RecordReader {
return array.ReaderFromIter(arrowSchema, func(yield func(arrow.RecordBatch, error) bool) {
bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
defer bldr.Release()
@@ -117,7 +237,12 @@ func (i InspectTable) manifestEntryReader(
_ = yield(nil, err)
}
- for _, manifest := range manifests {
+ for manifest, sourceErr := range source {
+ if sourceErr != nil {
+ yieldError(sourceErr)
+
+ return
+ }
if err := ctx.Err(); err != nil {
yieldError(err)
@@ -155,7 +280,7 @@ func (i InspectTable) manifestEntryReader(
} else if !emitted {
emitEmpty()
}
- }), nil
+ })
}
func emptyInspectRecordBatch(alloc memory.Allocator, schema *arrow.Schema) iter.Seq2[arrow.RecordBatch, error] {
@@ -266,13 +391,33 @@ func isInspectUnknownTransform(transform iceberg.Transform) bool {
}
}
-// DataFilesSchema returns the common content-file schema used by the data_files
-// and delete_files metadata tables. The partition field is omitted for an
-// unpartitioned table, as required by the Iceberg metadata-table spec.
+// DataFilesSchema returns the common content-file schema used by the files
+// metadata tables. The partition field is omitted for an unpartitioned table,
+// as required by the Iceberg metadata-table spec.
func DataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema {
return iceberg.NewSchema(0, inspectContentFileFields(partitionType)...)
}
+// FilesSchema returns the schema shared by the files metadata tables.
+func FilesSchema(partitionType *iceberg.StructType) *iceberg.Schema {
+ return DataFilesSchema(partitionType)
+}
+
+// AllFilesSchema returns the schema of the all_files metadata table.
+func AllFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema {
+ return DataFilesSchema(partitionType)
+}
+
+// AllDataFilesSchema returns the schema of the all_data_files metadata table.
+func AllDataFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema {
+ return DataFilesSchema(partitionType)
+}
+
+// AllDeleteFilesSchema returns the schema of the all_delete_files metadata table.
+func AllDeleteFilesSchema(partitionType *iceberg.StructType) *iceberg.Schema {
+ return DataFilesSchema(partitionType)
+}
+
const (
inspectContentFieldIDContent = 134
inspectContentFieldIDFilePath = 100
diff --git a/table/inspect_internal_test.go b/table/inspect_internal_test.go
index 71e75abe4..2444cf700 100644
--- a/table/inspect_internal_test.go
+++ b/table/inspect_internal_test.go
@@ -1686,6 +1686,263 @@ func inspectDataFileEntries(t *testing.T, spec iceberg.PartitionSpec, count int)
return entries
}
+func writeInspectManifest(
+ t *testing.T,
+ fs iceio.WriteFileIO,
+ path string,
+ spec iceberg.PartitionSpec,
+ schema *iceberg.Schema,
+ snapshotID int64,
+ content iceberg.ManifestContent,
+ entries []iceberg.ManifestEntry,
+) iceberg.ManifestFile {
+ t.Helper()
+
+ var buf bytes.Buffer
+ writer, err := iceberg.NewManifestWriter(2, &buf, spec, schema, snapshotID,
+ iceberg.WithManifestWriterContent(content))
+ require.NoError(t, err)
+ for _, entry := range entries {
+ switch entry.Status() {
+ case iceberg.EntryStatusDELETED:
+ require.NoError(t, writer.Delete(entry))
+ case iceberg.EntryStatusEXISTING:
+ require.NoError(t, writer.Existing(entry))
+ default:
+ require.NoError(t, writer.Add(entry))
+ }
+ }
+ require.NoError(t, writer.Close())
+ manifest, err := writer.ToManifestFile(path, int64(buf.Len()),
+ iceberg.WithManifestFileContent(content))
+ require.NoError(t, err)
+ require.NoError(t, fs.WriteFile(path, buf.Bytes()))
+
+ return manifest
+}
+
+func inspectAllFilesTable(t *testing.T) *Table {
+ t.Helper()
+
+ spec := *iceberg.UnpartitionedSpec
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ sequenceOne, sequenceTwo := int64(1), int64(2)
+ snapshotOne, snapshotTwo := int64(1), int64(2)
+
+ entry := func(snapshotID, sequenceNumber int64, file iceberg.DataFile) iceberg.ManifestEntry {
+ return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID,
+ &sequenceNumber, &sequenceNumber, file)
+ }
+ sharedData := newTestDataFile(t, spec, "mem://default/table-location/data/shared.parquet", nil)
+ deletedData := newTestDataFile(t, spec, "mem://default/table-location/data/deleted.parquet", nil)
+ newData := newTestDataFile(t, spec, "mem://default/table-location/data/new.parquet", nil)
+ deleteFile := newTestPosDeleteFileForSpec(t, spec,
+ "mem://default/table-location/data/delete.parquet", nil, sharedData.FilePath())
+ deletedSequence := sequenceOne
+ deletedEntry := iceberg.NewManifestEntry(iceberg.EntryStatusDELETED, &snapshotOne,
+ &deletedSequence, &deletedSequence, deletedData)
+
+ sharedManifest := writeInspectManifest(t, memIO,
+ "mem://default/table-location/metadata/shared.avro", spec, schema, snapshotOne,
+ iceberg.ManifestContentData, []iceberg.ManifestEntry{
+ entry(snapshotOne, sequenceOne, sharedData), deletedEntry,
+ })
+ // Keep a second manifest with the same file row to pin that all_* tables
+ // deduplicate shared manifests, but do not deduplicate rows across distinct
+ // manifests.
+ duplicateManifest := writeInspectManifest(t, memIO,
+ "mem://default/table-location/metadata/duplicate.avro", spec, schema, snapshotOne,
+ iceberg.ManifestContentData, []iceberg.ManifestEntry{entry(snapshotOne, sequenceOne, sharedData)})
+ newManifest := writeInspectManifest(t, memIO,
+ "mem://default/table-location/metadata/new.avro", spec, schema, snapshotTwo,
+ iceberg.ManifestContentData, []iceberg.ManifestEntry{entry(snapshotTwo, sequenceTwo, newData)})
+ deleteManifest := writeInspectManifest(t, memIO,
+ "mem://default/table-location/metadata/delete.avro", spec, schema, snapshotTwo,
+ iceberg.ManifestContentDeletes, []iceberg.ManifestEntry{entry(snapshotTwo, sequenceTwo, deleteFile)})
+
+ writeList := func(path string, snapshotID int64, parent *int64, sequenceNumber int64,
+ manifests []iceberg.ManifestFile,
+ ) []iceberg.ManifestFile {
+ var buf bytes.Buffer
+ require.NoError(t, iceberg.WriteManifestList(2, &buf, snapshotID, parent,
+ &sequenceNumber, 0, manifests))
+ require.NoError(t, memIO.WriteFile(path, buf.Bytes()))
+ written, err := iceberg.ReadManifestList(bytes.NewReader(buf.Bytes()))
+ require.NoError(t, err)
+
+ return written
+ }
+ listOne := "mem://default/table-location/metadata/snap-1-manifest-list.avro"
+ listTwo := "mem://default/table-location/metadata/snap-2-manifest-list.avro"
+ writtenOne := writeList(listOne, snapshotOne, nil, sequenceOne,
+ []iceberg.ManifestFile{sharedManifest, duplicateManifest})
+ writeList(listTwo, snapshotTwo, &snapshotOne, sequenceTwo,
+ []iceberg.ManifestFile{writtenOne[0], newManifest, deleteManifest})
+
+ txn.meta.snapshotList = []Snapshot{
+ {SnapshotID: snapshotOne, ManifestList: listOne, SequenceNumber: sequenceOne},
+ {SnapshotID: snapshotTwo, ParentSnapshotID: &snapshotOne, ManifestList: listTwo, SequenceNumber: sequenceTwo},
+ }
+ txn.meta.currentSnapshotID = &snapshotTwo
+ built, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ return New(Identifier{"db", "tbl"}, built, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return memIO, nil }, nil)
+}
+
+func inspectFileRows(t *testing.T, rr array.RecordReader) (paths []string, contents []int32) {
+ t.Helper()
+ defer rr.Release()
+ for rr.Next() {
+ record := rr.RecordBatch()
+ content := record.Column(0).(*array.Int32)
+ filePath := record.Column(1).(*array.String)
+ for row := range int(record.NumRows()) {
+ contents = append(contents, content.Value(row))
+ paths = append(paths, filePath.Value(row))
+ }
+ }
+ require.NoError(t, rr.Err())
+
+ return paths, contents
+}
+
+func TestInspectFilesTables(t *testing.T) {
+ tbl := inspectAllFilesTable(t)
+ tests := []struct {
+ name string
+ read func(context.Context) (array.RecordReader, error)
+ wantPaths []string
+ wantContent []int32
+ }{
+ {
+ name: "files",
+ read: tbl.Inspect().Files,
+ wantPaths: []string{
+ "mem://default/table-location/data/shared.parquet",
+ "mem://default/table-location/data/new.parquet",
+ "mem://default/table-location/data/delete.parquet",
+ },
+ wantContent: []int32{
+ int32(iceberg.EntryContentData), int32(iceberg.EntryContentData),
+ int32(iceberg.EntryContentPosDeletes),
+ },
+ },
+ {
+ name: "all files deduplicates shared manifest and preserves duplicate rows",
+ read: tbl.Inspect().AllFiles,
+ wantPaths: []string{
+ "mem://default/table-location/data/shared.parquet",
+ "mem://default/table-location/data/shared.parquet",
+ "mem://default/table-location/data/new.parquet",
+ "mem://default/table-location/data/delete.parquet",
+ },
+ wantContent: []int32{
+ int32(iceberg.EntryContentData), int32(iceberg.EntryContentData),
+ int32(iceberg.EntryContentData),
+ int32(iceberg.EntryContentPosDeletes),
+ },
+ },
+ {
+ name: "all data files",
+ read: tbl.Inspect().AllDataFiles,
+ wantPaths: []string{
+ "mem://default/table-location/data/shared.parquet",
+ "mem://default/table-location/data/shared.parquet",
+ "mem://default/table-location/data/new.parquet",
+ },
+ wantContent: []int32{
+ int32(iceberg.EntryContentData), int32(iceberg.EntryContentData),
+ int32(iceberg.EntryContentData),
+ },
+ },
+ {
+ name: "all delete files",
+ read: tbl.Inspect().AllDeleteFiles,
+ wantPaths: []string{"mem://default/table-location/data/delete.parquet"},
+ wantContent: []int32{int32(iceberg.EntryContentPosDeletes)},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ rr, err := tt.read(context.Background())
+ require.NoError(t, err)
+ paths, contents := inspectFileRows(t, rr)
+ require.Equal(t, tt.wantPaths, paths)
+ require.Equal(t, tt.wantContent, contents)
+ })
+ }
+}
+
+func TestInspectAllFilesStopsOnContextCancellation(t *testing.T) {
+ tbl := inspectAllFilesTable(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ rr, err := tbl.Inspect().AllFiles(ctx)
+ require.NoError(t, err)
+
+ cancel()
+ require.False(t, rr.Next())
+ require.ErrorIs(t, rr.Err(), context.Canceled)
+ rr.Release()
+}
+
+func TestInspectFilesTablesEarlyRelease(t *testing.T) {
+ checked := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ t.Cleanup(func() { checked.AssertSize(t, 0) })
+ tbl := inspectAllFilesTable(t)
+ reads := []struct {
+ name string
+ read func(context.Context) (array.RecordReader, error)
+ }{
+ {name: "files", read: tbl.Inspect(WithInspectAllocator(checked)).Files},
+ {name: "data files", read: tbl.Inspect(WithInspectAllocator(checked)).DataFiles},
+ {name: "delete files", read: tbl.Inspect(WithInspectAllocator(checked)).DeleteFiles},
+ {name: "all files", read: tbl.Inspect(WithInspectAllocator(checked)).AllFiles},
+ {name: "all data files", read: tbl.Inspect(WithInspectAllocator(checked)).AllDataFiles},
+ {name: "all delete files", read: tbl.Inspect(WithInspectAllocator(checked)).AllDeleteFiles},
+ }
+
+ for _, tt := range reads {
+ t.Run(tt.name, func(t *testing.T) {
+ rr, err := tt.read(context.Background())
+ require.NoError(t, err)
+ require.True(t, rr.Next())
+ rr.Release()
+ })
+ }
+}
+
+func TestInspectAllFilesReadsManifestListsLazily(t *testing.T) {
+ tbl := historyTestTable()
+ fs := iceio.NewMemFS()
+ tbl.fsF = func(context.Context) (iceio.IO, error) { return fs, nil }
+
+ rr, err := tbl.Inspect().AllFiles(context.Background())
+ require.NoError(t, err)
+ require.False(t, rr.Next())
+ require.ErrorContains(t, rr.Err(), "read snapshot 101 manifests")
+ rr.Release()
+}
+
+func TestInspectAllFilesSchemasMatchFiles(t *testing.T) {
+ partitionType := &iceberg.StructType{FieldList: []iceberg.NestedField{
+ {ID: 1000, Name: "part", Type: iceberg.PrimitiveTypes.String, Required: false},
+ }}
+ want := DataFilesSchema(partitionType)
+ for _, schema := range []*iceberg.Schema{
+ FilesSchema(partitionType),
+ AllFilesSchema(partitionType),
+ AllDataFilesSchema(partitionType),
+ AllDeleteFilesSchema(partitionType),
+ } {
+ require.True(t, want.Equals(schema))
+ }
+}
+
func TestInspectDataFilesStreamsBatchesAndSkipsDeleted(t *testing.T) {
const snapshotID = int64(1)
spec := *iceberg.UnpartitionedSpec
@@ -1786,11 +2043,26 @@ func TestInspectDataFilesEmitsEmptyBatchWhenAllEntriesAreDeleted(t *testing.T) {
rr, err := tbl.Inspect().DataFiles(context.Background())
require.NoError(t, err)
- defer rr.Release()
-
- record := collectRecord(t, rr)
- defer record.Release()
+ require.True(t, rr.Next())
+ record := rr.RecordBatch()
require.EqualValues(t, 0, record.NumRows())
+ rr.Release()
+}
+
+func TestInspectDataFilesEmptyTableEarlyRelease(t *testing.T) {
+ meta := &metadataV2{commonMetadata: commonMetadata{
+ SchemaList: []*iceberg.Schema{simpleSchema()},
+ CurrentSchemaID: 0,
+ Specs: []iceberg.PartitionSpec{*iceberg.UnpartitionedSpec},
+ DefaultSpecID: 0,
+ }}
+ tbl := New(Identifier{"empty-data-files"}, meta, "metadata.json", nil, nil)
+
+ rr, err := tbl.Inspect().DataFiles(context.Background())
+ require.NoError(t, err)
+ require.True(t, rr.Next())
+ require.EqualValues(t, 0, rr.RecordBatch().NumRows())
+ rr.Release()
}
func TestInspectDataFilesAllocator(t *testing.T) {