laskoviymishka commented on code in PR #1697:
URL: https://github.com/apache/iceberg-go/pull/1697#discussion_r3749988029
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
Review Comment:
For a read-only inspect API I'm a little wary of hard-erroring the whole
call when we hit a single manifest with a content value we don't recognize —
one forward-version manifest makes the entire table uninspectable through this
method. Java/PyIceberg lean toward graceful degradation here.
Not blocking, and I can see the case for failing loud. But I'd at least
consider skip-with-warning or a wrapped sentinel the caller can pick out. Same
thought applies to the excess-summaries reject a few lines down. wdyt?
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
+ }
+
+ content.Append(int32(manifestContent))
+ path.Append(manifest.FilePath())
+ length.Append(manifest.Length())
+ partitionSpecID.Append(manifest.PartitionSpecID())
+ addedSnapshotID.Append(manifest.SnapshotID())
Review Comment:
`added_snapshot_id` is declared `Required:true`, but `SnapshotID()` returns
`-1` for a V1 manifest whose `added_snapshot_id` field was absent, and we
append it straight through. Java returns null there, so a query joining
`manifests.added_snapshot_id` against the snapshots table would silently miss
instead of seeing null, and `-1` reads like a real id.
I don't think any of the current constructors can actually produce the `-1`
(`NewManifestFile` always injects a concrete id), so this may not be reachable
today — but if the Avro read path can surface it, I'd declare the field
`Required:false` and `AppendNull()` when `SnapshotID()==-1`. Worth confirming
which way that goes.
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
+ }
+
+ content.Append(int32(manifestContent))
+ path.Append(manifest.FilePath())
+ length.Append(manifest.Length())
+ partitionSpecID.Append(manifest.PartitionSpecID())
+ addedSnapshotID.Append(manifest.SnapshotID())
+ appendCount := func(builder *array.Int32Builder, name string,
count int32) error {
+ if err := appendManifestCount(builder,
manifest.Version(), name, count); err != nil {
+ return fmt.Errorf("manifest %s: %w",
manifest.FilePath(), err)
+ }
+
+ return nil
+ }
+
+ switch manifestContent {
+ case iceberg.ManifestContentData:
+ if err := appendCount(addedDataFiles,
"added_data_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDataFiles,
"existing_data_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDataFiles,
"deleted_data_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ addedDeleteFiles.Append(0)
+ existingDeleteFiles.Append(0)
+ deletedDeleteFiles.Append(0)
+ case iceberg.ManifestContentDeletes:
+ addedDataFiles.Append(0)
+ existingDataFiles.Append(0)
+ deletedDataFiles.Append(0)
+ if err := appendCount(addedDeleteFiles,
"added_delete_files", manifest.AddedDataFiles()); err != nil {
Review Comment:
This routing is going to make the next reader do a double-take: for delete
manifests we're pulling the delete-file counts out of
`AddedDataFiles()`/`ExistingDataFiles()`/`DeletedDataFiles()`. It's correct
today because those accessors return the generic `added_files_count` field,
which holds delete-file counts for a delete manifest — but the names say the
opposite.
A one-line comment noting that would save the confusion, and flag it if
iceberg-go ever grows distinct `AddedDeleteFiles()` accessors.
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
+ }
+
+ content.Append(int32(manifestContent))
+ path.Append(manifest.FilePath())
+ length.Append(manifest.Length())
+ partitionSpecID.Append(manifest.PartitionSpecID())
+ addedSnapshotID.Append(manifest.SnapshotID())
+ appendCount := func(builder *array.Int32Builder, name string,
count int32) error {
+ if err := appendManifestCount(builder,
manifest.Version(), name, count); err != nil {
+ return fmt.Errorf("manifest %s: %w",
manifest.FilePath(), err)
+ }
+
+ return nil
+ }
+
+ switch manifestContent {
+ case iceberg.ManifestContentData:
+ if err := appendCount(addedDataFiles,
"added_data_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDataFiles,
"existing_data_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDataFiles,
"deleted_data_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ addedDeleteFiles.Append(0)
+ existingDeleteFiles.Append(0)
+ deletedDeleteFiles.Append(0)
+ case iceberg.ManifestContentDeletes:
+ addedDataFiles.Append(0)
+ existingDataFiles.Append(0)
+ deletedDataFiles.Append(0)
+ if err := appendCount(addedDeleteFiles,
"added_delete_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDeleteFiles,
"existing_delete_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDeleteFiles,
"deleted_delete_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ }
+
+ spec :=
i.tbl.metadata.PartitionSpecByID(int(manifest.PartitionSpecID()))
+ if spec == nil {
Review Comment:
I think there's a subtle ordering bug here. We look up the spec and error
out if it's nil before we check `partitions == nil`, so a manifest that carries
no partition summaries but was written under a spec that's since been removed
(spec evolution + metadata cleanup) fails the whole `Manifests()` call.
Java and PyIceberg return early on the no-summaries case and never touch the
spec, so they succeed where we'd error. I'd move the `partitions == nil` check
above the spec lookup — that also gives a clean home for the empty-list change
from the other thread. wdyt?
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
+ }
+
+ content.Append(int32(manifestContent))
+ path.Append(manifest.FilePath())
+ length.Append(manifest.Length())
+ partitionSpecID.Append(manifest.PartitionSpecID())
+ addedSnapshotID.Append(manifest.SnapshotID())
+ appendCount := func(builder *array.Int32Builder, name string,
count int32) error {
+ if err := appendManifestCount(builder,
manifest.Version(), name, count); err != nil {
+ return fmt.Errorf("manifest %s: %w",
manifest.FilePath(), err)
+ }
+
+ return nil
+ }
+
+ switch manifestContent {
+ case iceberg.ManifestContentData:
+ if err := appendCount(addedDataFiles,
"added_data_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDataFiles,
"existing_data_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDataFiles,
"deleted_data_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ addedDeleteFiles.Append(0)
+ existingDeleteFiles.Append(0)
+ deletedDeleteFiles.Append(0)
+ case iceberg.ManifestContentDeletes:
+ addedDataFiles.Append(0)
+ existingDataFiles.Append(0)
+ deletedDataFiles.Append(0)
+ if err := appendCount(addedDeleteFiles,
"added_delete_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDeleteFiles,
"existing_delete_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDeleteFiles,
"deleted_delete_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ }
+
+ spec :=
i.tbl.metadata.PartitionSpecByID(int(manifest.PartitionSpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("manifest %s references missing
partition spec %d",
+ manifest.FilePath(), manifest.PartitionSpecID())
+ }
+ partType := spec.PartitionType(i.tbl.metadata.CurrentSchema())
+ partitions := manifest.Partitions()
+ if partitions == nil {
+ partitionSummaries.AppendNull()
+
+ continue
+ }
+ if len(partitions) > spec.NumFields() {
+ return nil, fmt.Errorf("manifest %s has %d partition
summaries for partition spec %d with %d fields",
+ manifest.FilePath(), len(partitions),
manifest.PartitionSpecID(), spec.NumFields())
+ }
+
+ partitionSummaries.Append(true)
+ for idx, summary := range partitions {
+ summaryStruct.Append(true)
+ summaryContainsNull.Append(summary.ContainsNull)
+ if summary.ContainsNaN == nil {
+ summaryContainsNaN.AppendNull()
+ } else {
+ summaryContainsNaN.Append(*summary.ContainsNaN)
+ }
+
+ fieldType := partType.FieldList[idx].Type
+ transform := spec.Field(idx).Transform
+ if err := appendManifestBound(summaryLower, fieldType,
transform, summary.LowerBound); err != nil {
+ return nil, fmt.Errorf("manifest %s partition
field %d lower bound: %w", manifest.FilePath(), idx, err)
+ }
+ if err := appendManifestBound(summaryUpper, fieldType,
transform, summary.UpperBound); err != nil {
+ return nil, fmt.Errorf("manifest %s partition
field %d upper bound: %w", manifest.FilePath(), idx, err)
+ }
+ }
+ }
+
+ rr, err := singleBatchReader(arrowSchema, bldr)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ return rr, nil
+}
+
+func appendManifestCount(builder *array.Int32Builder, version int, name
string, count int32) error {
+ if count < 0 {
+ if version == 1 {
+ builder.AppendNull()
Review Comment:
This is the one I'd block on. The schema declares the six count fields
`Required:true`, which `SchemaToArrowSchema` turns into Arrow `Nullable:false`
— but here we `AppendNull()` for V1 manifests, so we emit a batch whose schema
says non-nullable while the data carries nulls. Arrow-Go won't panic on that,
but any consumer that trusts the schema (a Parquet writer, a scan engine) reads
it as corrupt.
Every reference client is internally consistent and none emit null here:
PyIceberg and Java declare the fields and write 0 for V1 unknowns; Rust
declares them nullable. I'd match PyIceberg/Java and `Append(int32(0))` instead
of `AppendNull()` — that keeps `Required:true` honest. If we'd rather preserve
the null, the fields (5, 6, 7, 15, 16, 17) have to flip to `Required:false`.
Either is fine, but the schema and the data have to agree.
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
+ }
+
+ content.Append(int32(manifestContent))
+ path.Append(manifest.FilePath())
+ length.Append(manifest.Length())
+ partitionSpecID.Append(manifest.PartitionSpecID())
+ addedSnapshotID.Append(manifest.SnapshotID())
+ appendCount := func(builder *array.Int32Builder, name string,
count int32) error {
+ if err := appendManifestCount(builder,
manifest.Version(), name, count); err != nil {
+ return fmt.Errorf("manifest %s: %w",
manifest.FilePath(), err)
+ }
+
+ return nil
+ }
+
+ switch manifestContent {
+ case iceberg.ManifestContentData:
+ if err := appendCount(addedDataFiles,
"added_data_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDataFiles,
"existing_data_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDataFiles,
"deleted_data_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ addedDeleteFiles.Append(0)
+ existingDeleteFiles.Append(0)
+ deletedDeleteFiles.Append(0)
+ case iceberg.ManifestContentDeletes:
+ addedDataFiles.Append(0)
+ existingDataFiles.Append(0)
+ deletedDataFiles.Append(0)
+ if err := appendCount(addedDeleteFiles,
"added_delete_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDeleteFiles,
"existing_delete_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDeleteFiles,
"deleted_delete_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ }
+
+ spec :=
i.tbl.metadata.PartitionSpecByID(int(manifest.PartitionSpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("manifest %s references missing
partition spec %d",
+ manifest.FilePath(), manifest.PartitionSpecID())
+ }
+ partType := spec.PartitionType(i.tbl.metadata.CurrentSchema())
+ partitions := manifest.Partitions()
+ if partitions == nil {
+ partitionSummaries.AppendNull()
+
+ continue
+ }
+ if len(partitions) > spec.NumFields() {
+ return nil, fmt.Errorf("manifest %s has %d partition
summaries for partition spec %d with %d fields",
+ manifest.FilePath(), len(partitions),
manifest.PartitionSpecID(), spec.NumFields())
+ }
+
+ partitionSummaries.Append(true)
+ for idx, summary := range partitions {
+ summaryStruct.Append(true)
+ summaryContainsNull.Append(summary.ContainsNull)
+ if summary.ContainsNaN == nil {
+ summaryContainsNaN.AppendNull()
+ } else {
+ summaryContainsNaN.Append(*summary.ContainsNaN)
+ }
+
+ fieldType := partType.FieldList[idx].Type
+ transform := spec.Field(idx).Transform
+ if err := appendManifestBound(summaryLower, fieldType,
transform, summary.LowerBound); err != nil {
+ return nil, fmt.Errorf("manifest %s partition
field %d lower bound: %w", manifest.FilePath(), idx, err)
Review Comment:
nit: this reports the partition field by positional index ("partition field
3"), which makes the user count schema fields to find it. We've got
`partType.FieldList[idx].Name` right here — worth using the name instead.
##########
table/inspect_internal_test.go:
##########
@@ -404,6 +457,308 @@ func TestInspectSnapshotsEmpty(t *testing.T) {
require.EqualValues(t, 6, rec.NumCols())
}
+func TestInspectManifestsSchema(t *testing.T) {
+ sc := ManifestsSchema()
+
+ require.Equal(t,
+ []string{
+ "content", "path", "length", "partition_spec_id",
"added_snapshot_id",
+ "added_data_files_count", "existing_data_files_count",
"deleted_data_files_count",
+ "added_delete_files_count",
"existing_delete_files_count", "deleted_delete_files_count",
+ "partition_summaries",
+ },
+ testFieldNames(sc))
+
+ fields := sc.Fields()
+ require.Equal(t, 14, fields[0].ID)
+ require.Equal(t, 1, fields[1].ID)
+ require.Equal(t, 17, fields[10].ID)
+ require.Equal(t, 8, fields[11].ID)
+ require.True(t, fields[11].Required)
+ require.True(t, fields[11].Type.(*iceberg.ListType).ElementRequired)
+ require.Equal(t, 9, fields[11].Type.(*iceberg.ListType).ElementID)
+
+ partitionSummary :=
fields[11].Type.(*iceberg.ListType).Element.(*iceberg.StructType)
+ require.Equal(t,
+ []string{"contains_null", "contains_nan", "lower_bound",
"upper_bound"},
+ testFieldNames(iceberg.NewSchema(0,
partitionSummary.FieldList...)))
+ require.Equal(t, 10, partitionSummary.FieldList[0].ID)
+ require.Equal(t, 11, partitionSummary.FieldList[1].ID)
+ require.Equal(t, 12, partitionSummary.FieldList[2].ID)
+ require.Equal(t, 13, partitionSummary.FieldList[3].ID)
+}
+
+func TestInspectManifests(t *testing.T) {
+ const snapshotID = int64(1)
+ spec := partitionedSpec()
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ file := newTestDataFileWithCount(t, spec,
+ "mem://default/table-location/data/data.parquet",
map[int]any{1000: int32(7)}, 3)
+ sequenceNumber := int64(1)
+ entry := iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, int64Ptr(snapshotID),
&sequenceNumber, &sequenceNumber, file)
+
+ manifestPath :=
"mem://default/table-location/metadata/data-manifest.avro"
+ manifestListPath :=
"mem://default/table-location/metadata/snap-1-manifest-list.avro"
+ var manifestBuf bytes.Buffer
+ manifest, err := iceberg.WriteManifest(manifestPath, &manifestBuf, 2,
spec, schema, snapshotID,
+ []iceberg.ManifestEntry{entry})
+ require.NoError(t, err)
+ require.NoError(t, memIO.WriteFile(manifestPath, manifestBuf.Bytes()))
+
+ var listBuf bytes.Buffer
+ require.NoError(t, iceberg.WriteManifestList(2, &listBuf, snapshotID,
nil, &sequenceNumber, 0,
+ []iceberg.ManifestFile{manifest}))
+ require.NoError(t, memIO.WriteFile(manifestListPath, listBuf.Bytes()))
+
+ snapID := snapshotID
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: snapshotID,
+ ManifestList: manifestListPath,
+ SequenceNumber: sequenceNumber,
+ }}
+ txn.meta.currentSnapshotID = &snapID
+ built, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"db", "tbl"}, built, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return memIO, nil },
nil)
+ rr, err := tbl.Inspect().Manifests(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ record := collectRecord(t, rr)
+ defer record.Release()
+ require.EqualValues(t, 1, record.NumRows())
+ require.EqualValues(t, 1, record.Column(5).(*array.Int32).Value(0))
+ require.EqualValues(t, 0, record.Column(8).(*array.Int32).Value(0))
+ require.Equal(t, manifestPath,
record.Column(1).(*array.String).Value(0))
+
+ summaries := record.Column(11).(*array.List)
+ require.False(t, summaries.IsNull(0))
+ start, end := summaries.ValueOffsets(0)
+ require.EqualValues(t, 1, end-start)
+ summary := summaries.ListValues().(*array.Struct)
+ require.False(t, summary.Field(0).(*array.Boolean).Value(0))
+ require.Equal(t, "7", summary.Field(2).(*array.String).Value(0))
Review Comment:
The non-null `contains_nan` path never gets exercised. Every test leaves
`ContainsNaN` nil, so we only ever hit the `AppendNull` branch — the
`Append(*summary.ContainsNaN)` side would pass just as well if it were deleted.
This block also skips asserting `Field(1)` entirely.
I'd add a case with `ContainsNaN` set to a bool and assert `Field(1)` is
non-null with the right value.
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
+ }
+
+ content.Append(int32(manifestContent))
+ path.Append(manifest.FilePath())
+ length.Append(manifest.Length())
+ partitionSpecID.Append(manifest.PartitionSpecID())
+ addedSnapshotID.Append(manifest.SnapshotID())
+ appendCount := func(builder *array.Int32Builder, name string,
count int32) error {
+ if err := appendManifestCount(builder,
manifest.Version(), name, count); err != nil {
+ return fmt.Errorf("manifest %s: %w",
manifest.FilePath(), err)
+ }
+
+ return nil
+ }
+
+ switch manifestContent {
+ case iceberg.ManifestContentData:
+ if err := appendCount(addedDataFiles,
"added_data_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDataFiles,
"existing_data_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDataFiles,
"deleted_data_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ addedDeleteFiles.Append(0)
+ existingDeleteFiles.Append(0)
+ deletedDeleteFiles.Append(0)
+ case iceberg.ManifestContentDeletes:
+ addedDataFiles.Append(0)
+ existingDataFiles.Append(0)
+ deletedDataFiles.Append(0)
+ if err := appendCount(addedDeleteFiles,
"added_delete_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDeleteFiles,
"existing_delete_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDeleteFiles,
"deleted_delete_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ }
+
+ spec :=
i.tbl.metadata.PartitionSpecByID(int(manifest.PartitionSpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("manifest %s references missing
partition spec %d",
+ manifest.FilePath(), manifest.PartitionSpecID())
+ }
+ partType := spec.PartitionType(i.tbl.metadata.CurrentSchema())
+ partitions := manifest.Partitions()
+ if partitions == nil {
+ partitionSummaries.AppendNull()
Review Comment:
Same contract problem as the count columns, one column over.
`partition_summaries` is field 8, `Required:true` → `Nullable:false`, but when
`partitions == nil` we `AppendNull()` into it. PyIceberg and Rust both emit an
empty list here, never null.
I'd emit `partitionSummaries.Append(true)` with no children (a zero-length
list), then `continue`. One gotcha: `Append(false)` would still be a null slot,
so it has to be `Append(true)` to get a non-null empty list. That matches the
reference clients and keeps the column honestly non-nullable.
##########
table/inspect.go:
##########
@@ -247,6 +248,224 @@ func (i InspectTable) Snapshots(ctx context.Context)
(array.RecordReader, error)
return rr, nil
}
+// Manifests returns one row for each manifest in the current snapshot.
+// Partition summaries are exposed as the human-readable values used by the
+// other Iceberg clients.
+func (i InspectTable) Manifests(ctx context.Context) (array.RecordReader,
error) {
+ schema := ManifestsSchema()
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: build arrow schema:
%w", err)
+ }
+
+ manifests, err := i.currentSnapshotManifests(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ bldr := array.NewRecordBuilder(i.alloc, arrowSchema)
+ defer bldr.Release()
+
+ content := bldr.Field(0).(*array.Int32Builder)
+ path := bldr.Field(1).(*array.StringBuilder)
+ length := bldr.Field(2).(*array.Int64Builder)
+ partitionSpecID := bldr.Field(3).(*array.Int32Builder)
+ addedSnapshotID := bldr.Field(4).(*array.Int64Builder)
+ addedDataFiles := bldr.Field(5).(*array.Int32Builder)
+ existingDataFiles := bldr.Field(6).(*array.Int32Builder)
+ deletedDataFiles := bldr.Field(7).(*array.Int32Builder)
+ addedDeleteFiles := bldr.Field(8).(*array.Int32Builder)
+ existingDeleteFiles := bldr.Field(9).(*array.Int32Builder)
+ deletedDeleteFiles := bldr.Field(10).(*array.Int32Builder)
+ partitionSummaries := bldr.Field(11).(*array.ListBuilder)
+ summaryStruct :=
partitionSummaries.ValueBuilder().(*array.StructBuilder)
+ summaryContainsNull :=
summaryStruct.FieldBuilder(0).(*array.BooleanBuilder)
+ summaryContainsNaN :=
summaryStruct.FieldBuilder(1).(*array.BooleanBuilder)
+ summaryLower := summaryStruct.FieldBuilder(2).(*array.StringBuilder)
+ summaryUpper := summaryStruct.FieldBuilder(3).(*array.StringBuilder)
+
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+
+ manifestContent := manifest.ManifestContent()
+ switch manifestContent {
+ case iceberg.ManifestContentData,
iceberg.ManifestContentDeletes:
+ default:
+ return nil, fmt.Errorf("manifest %s has unknown content
%d", manifest.FilePath(), manifestContent)
+ }
+
+ content.Append(int32(manifestContent))
+ path.Append(manifest.FilePath())
+ length.Append(manifest.Length())
+ partitionSpecID.Append(manifest.PartitionSpecID())
+ addedSnapshotID.Append(manifest.SnapshotID())
+ appendCount := func(builder *array.Int32Builder, name string,
count int32) error {
+ if err := appendManifestCount(builder,
manifest.Version(), name, count); err != nil {
+ return fmt.Errorf("manifest %s: %w",
manifest.FilePath(), err)
+ }
+
+ return nil
+ }
+
+ switch manifestContent {
+ case iceberg.ManifestContentData:
+ if err := appendCount(addedDataFiles,
"added_data_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDataFiles,
"existing_data_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDataFiles,
"deleted_data_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ addedDeleteFiles.Append(0)
+ existingDeleteFiles.Append(0)
+ deletedDeleteFiles.Append(0)
+ case iceberg.ManifestContentDeletes:
+ addedDataFiles.Append(0)
+ existingDataFiles.Append(0)
+ deletedDataFiles.Append(0)
+ if err := appendCount(addedDeleteFiles,
"added_delete_files", manifest.AddedDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(existingDeleteFiles,
"existing_delete_files", manifest.ExistingDataFiles()); err != nil {
+ return nil, err
+ }
+ if err := appendCount(deletedDeleteFiles,
"deleted_delete_files", manifest.DeletedDataFiles()); err != nil {
+ return nil, err
+ }
+ }
+
+ spec :=
i.tbl.metadata.PartitionSpecByID(int(manifest.PartitionSpecID()))
+ if spec == nil {
+ return nil, fmt.Errorf("manifest %s references missing
partition spec %d",
+ manifest.FilePath(), manifest.PartitionSpecID())
+ }
+ partType := spec.PartitionType(i.tbl.metadata.CurrentSchema())
+ partitions := manifest.Partitions()
+ if partitions == nil {
+ partitionSummaries.AppendNull()
+
+ continue
+ }
+ if len(partitions) > spec.NumFields() {
+ return nil, fmt.Errorf("manifest %s has %d partition
summaries for partition spec %d with %d fields",
+ manifest.FilePath(), len(partitions),
manifest.PartitionSpecID(), spec.NumFields())
+ }
+
+ partitionSummaries.Append(true)
+ for idx, summary := range partitions {
+ summaryStruct.Append(true)
+ summaryContainsNull.Append(summary.ContainsNull)
+ if summary.ContainsNaN == nil {
+ summaryContainsNaN.AppendNull()
+ } else {
+ summaryContainsNaN.Append(*summary.ContainsNaN)
+ }
+
+ fieldType := partType.FieldList[idx].Type
+ transform := spec.Field(idx).Transform
+ if err := appendManifestBound(summaryLower, fieldType,
transform, summary.LowerBound); err != nil {
+ return nil, fmt.Errorf("manifest %s partition
field %d lower bound: %w", manifest.FilePath(), idx, err)
+ }
+ if err := appendManifestBound(summaryUpper, fieldType,
transform, summary.UpperBound); err != nil {
+ return nil, fmt.Errorf("manifest %s partition
field %d upper bound: %w", manifest.FilePath(), idx, err)
+ }
+ }
+ }
+
+ rr, err := singleBatchReader(arrowSchema, bldr)
+ if err != nil {
+ return nil, fmt.Errorf("inspect manifests: %w", err)
+ }
+
+ return rr, nil
+}
+
+func appendManifestCount(builder *array.Int32Builder, version int, name
string, count int32) error {
+ if count < 0 {
+ if version == 1 {
+ builder.AppendNull()
+
+ return nil
+ }
+
+ return fmt.Errorf("negative %s count %d in manifest list
version %d", name, count, version)
+ }
+
+ builder.Append(count)
+
+ return nil
+}
+
+func (i InspectTable) currentSnapshotManifests(ctx context.Context)
([]iceberg.ManifestFile, error) {
+ snapshot := i.tbl.metadata.CurrentSnapshot()
+ if snapshot == nil {
+ return nil, 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
Review Comment:
Small thing — this `err` comes back unwrapped while everything else in the
chain gets an `fmt.Errorf`. The outer wrap is just `"inspect manifests: %w"`,
so a failure in the IO factory loses the context that it happened here. I'd do
`return nil, fmt.Errorf("get file IO: %w", err)`.
##########
table/inspect_internal_test.go:
##########
@@ -404,6 +457,308 @@ func TestInspectSnapshotsEmpty(t *testing.T) {
require.EqualValues(t, 6, rec.NumCols())
}
+func TestInspectManifestsSchema(t *testing.T) {
+ sc := ManifestsSchema()
+
+ require.Equal(t,
+ []string{
+ "content", "path", "length", "partition_spec_id",
"added_snapshot_id",
+ "added_data_files_count", "existing_data_files_count",
"deleted_data_files_count",
+ "added_delete_files_count",
"existing_delete_files_count", "deleted_delete_files_count",
+ "partition_summaries",
+ },
+ testFieldNames(sc))
+
+ fields := sc.Fields()
+ require.Equal(t, 14, fields[0].ID)
+ require.Equal(t, 1, fields[1].ID)
+ require.Equal(t, 17, fields[10].ID)
+ require.Equal(t, 8, fields[11].ID)
+ require.True(t, fields[11].Required)
+ require.True(t, fields[11].Type.(*iceberg.ListType).ElementRequired)
+ require.Equal(t, 9, fields[11].Type.(*iceberg.ListType).ElementID)
+
+ partitionSummary :=
fields[11].Type.(*iceberg.ListType).Element.(*iceberg.StructType)
+ require.Equal(t,
+ []string{"contains_null", "contains_nan", "lower_bound",
"upper_bound"},
+ testFieldNames(iceberg.NewSchema(0,
partitionSummary.FieldList...)))
+ require.Equal(t, 10, partitionSummary.FieldList[0].ID)
+ require.Equal(t, 11, partitionSummary.FieldList[1].ID)
+ require.Equal(t, 12, partitionSummary.FieldList[2].ID)
+ require.Equal(t, 13, partitionSummary.FieldList[3].ID)
+}
+
+func TestInspectManifests(t *testing.T) {
+ const snapshotID = int64(1)
+ spec := partitionedSpec()
+ txn, memIO := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ file := newTestDataFileWithCount(t, spec,
+ "mem://default/table-location/data/data.parquet",
map[int]any{1000: int32(7)}, 3)
+ sequenceNumber := int64(1)
+ entry := iceberg.NewManifestEntry(
+ iceberg.EntryStatusADDED, int64Ptr(snapshotID),
&sequenceNumber, &sequenceNumber, file)
+
+ manifestPath :=
"mem://default/table-location/metadata/data-manifest.avro"
+ manifestListPath :=
"mem://default/table-location/metadata/snap-1-manifest-list.avro"
+ var manifestBuf bytes.Buffer
+ manifest, err := iceberg.WriteManifest(manifestPath, &manifestBuf, 2,
spec, schema, snapshotID,
+ []iceberg.ManifestEntry{entry})
+ require.NoError(t, err)
+ require.NoError(t, memIO.WriteFile(manifestPath, manifestBuf.Bytes()))
+
+ var listBuf bytes.Buffer
+ require.NoError(t, iceberg.WriteManifestList(2, &listBuf, snapshotID,
nil, &sequenceNumber, 0,
+ []iceberg.ManifestFile{manifest}))
+ require.NoError(t, memIO.WriteFile(manifestListPath, listBuf.Bytes()))
+
+ snapID := snapshotID
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: snapshotID,
+ ManifestList: manifestListPath,
+ SequenceNumber: sequenceNumber,
+ }}
+ txn.meta.currentSnapshotID = &snapID
+ built, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"db", "tbl"}, built, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return memIO, nil },
nil)
+ rr, err := tbl.Inspect().Manifests(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+
+ record := collectRecord(t, rr)
+ defer record.Release()
+ require.EqualValues(t, 1, record.NumRows())
+ require.EqualValues(t, 1, record.Column(5).(*array.Int32).Value(0))
+ require.EqualValues(t, 0, record.Column(8).(*array.Int32).Value(0))
+ require.Equal(t, manifestPath,
record.Column(1).(*array.String).Value(0))
+
+ summaries := record.Column(11).(*array.List)
+ require.False(t, summaries.IsNull(0))
+ start, end := summaries.ValueOffsets(0)
+ require.EqualValues(t, 1, end-start)
+ summary := summaries.ListValues().(*array.Struct)
+ require.False(t, summary.Field(0).(*array.Boolean).Value(0))
+ require.Equal(t, "7", summary.Field(2).(*array.String).Value(0))
+ require.Equal(t, "7", summary.Field(3).(*array.String).Value(0))
+}
+
+func TestInspectManifestsPromotedPartitionSummaryBounds(t *testing.T) {
+ tests := []struct {
+ name string
+ initialType iceberg.Type
+ currentType iceberg.Type
+ literal iceberg.Literal
+ expected string
+ }{
+ {
+ name: "int to long",
+ initialType: iceberg.PrimitiveTypes.Int32,
+ currentType: iceberg.PrimitiveTypes.Int64,
+ literal: iceberg.Int32Literal(7),
+ expected: "7",
+ },
+ {
+ name: "float to double",
+ initialType: iceberg.PrimitiveTypes.Float32,
+ currentType: iceberg.PrimitiveTypes.Float64,
+ literal: iceberg.Float32Literal(1.5),
+ expected: "1.5",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ spec := partitionedSpec()
+ initialSchema := iceberg.NewSchema(0,
iceberg.NestedField{
+ ID: 1, Name: "id", Type: tt.initialType,
Required: true,
+ })
+ currentSchema := iceberg.NewSchema(1,
iceberg.NestedField{
+ ID: 1, Name: "id", Type: tt.currentType,
Required: true,
+ })
+ bound, err := tt.literal.MarshalBinary()
+ require.NoError(t, err)
+ manifest := iceberg.NewManifestFile(2,
"mem://default/table-location/metadata/promoted.avro",
+ 100, int32(spec.ID()), 1).
+ SequenceNum(1, 1).
+ Partitions([]iceberg.FieldSummary{{LowerBound:
&bound, UpperBound: &bound}}).
+ Build()
+ tbl := inspectTableWithManifestListAndSchemas(t,
initialSchema, currentSchema,
+ spec, 2, []iceberg.ManifestFile{manifest})
+
+ rr, err := tbl.Inspect().Manifests(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+ record := collectRecord(t, rr)
+ defer record.Release()
+
+ summaries := record.Column(11).(*array.List)
+ start, end := summaries.ValueOffsets(0)
+ require.EqualValues(t, 1, end-start)
+ summary := summaries.ListValues().(*array.Struct)
+ require.Equal(t, tt.expected,
summary.Field(2).(*array.String).Value(0))
+ require.Equal(t, tt.expected,
summary.Field(3).(*array.String).Value(0))
+ })
+ }
+}
+
+func TestInspectManifestsDroppedPartitionSource(t *testing.T) {
+ spec := partitionedSpec()
+ initialSchema := simpleSchema()
+ currentSchema := iceberg.NewSchema(1)
+ bound, err := iceberg.Int32Literal(7).MarshalBinary()
+ require.NoError(t, err)
+ manifest := iceberg.NewManifestFile(2,
"mem://default/table-location/metadata/dropped-source.avro",
+ 100, int32(spec.ID()), 1).
+ SequenceNum(1, 1).
+ Partitions([]iceberg.FieldSummary{{LowerBound: &bound,
UpperBound: &bound}}).
+ Build()
+ tbl := inspectTableWithManifestListAndSchemas(t, initialSchema,
currentSchema,
+ spec, 2, []iceberg.ManifestFile{manifest})
+
+ rr, err := tbl.Inspect().Manifests(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+ record := collectRecord(t, rr)
+ defer record.Release()
+
+ summaries := record.Column(11).(*array.List)
+ start, end := summaries.ValueOffsets(0)
+ require.EqualValues(t, 1, end-start)
+ summary := summaries.ListValues().(*array.Struct)
+ require.True(t, summary.Field(2).(*array.String).IsNull(0))
+ require.True(t, summary.Field(3).(*array.String).IsNull(0))
+}
+
+func TestInspectManifestsDeleteCounts(t *testing.T) {
+ spec := partitionedSpec()
+ manifest := iceberg.NewManifestFile(2,
"mem://default/table-location/metadata/delete-manifest.avro",
+ 100, int32(spec.ID()), 1).
+ Content(iceberg.ManifestContentDeletes).
+ SequenceNum(1, 1).
+ AddedFiles(2).
+ ExistingFiles(3).
+ DeletedFiles(4).
+ Build()
+ tbl := inspectTableWithManifestList(t, spec, 2,
[]iceberg.ManifestFile{manifest})
+
+ rr, err := tbl.Inspect().Manifests(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+ record := collectRecord(t, rr)
+ defer record.Release()
+
+ require.EqualValues(t, iceberg.ManifestContentDeletes,
record.Column(0).(*array.Int32).Value(0))
+ for _, col := range []int{5, 6, 7} {
+ require.EqualValues(t, 0,
record.Column(col).(*array.Int32).Value(0))
+ }
+ require.EqualValues(t, 2, record.Column(8).(*array.Int32).Value(0))
+ require.EqualValues(t, 3, record.Column(9).(*array.Int32).Value(0))
+ require.EqualValues(t, 4, record.Column(10).(*array.Int32).Value(0))
+}
+
+func TestInspectManifestsV1UnknownCounts(t *testing.T) {
+ spec := partitionedSpec()
+ manifest := iceberg.NewManifestFile(1,
"mem://default/table-location/metadata/v1-manifest.avro",
+ 100, int32(spec.ID()), 1).
+ AddedFiles(-1).
+ ExistingFiles(-1).
+ DeletedFiles(-1).
+ Build()
+ tbl := inspectTableWithManifestList(t, spec, 1,
[]iceberg.ManifestFile{manifest})
+
+ rr, err := tbl.Inspect().Manifests(context.Background())
+ require.NoError(t, err)
+ defer rr.Release()
+ record := collectRecord(t, rr)
+ defer record.Release()
+
+ for _, col := range []int{5, 6, 7} {
+ require.True(t, record.Column(col).(*array.Int32).IsNull(0))
+ }
+ for _, col := range []int{8, 9, 10} {
+ require.EqualValues(t, 0,
record.Column(col).(*array.Int32).Value(0))
+ }
+ require.True(t, record.Column(11).(*array.List).IsNull(0))
+}
+
+func TestInspectManifestsRejectsNegativeCountsForV2AndV3(t *testing.T) {
+ tests := []struct {
+ name string
+ version int
+ added, existing, deleted int32
+ invalidCountName string
+ }{
+ {name: "v2 added data files", version: 2, added: -123,
existing: 1, deleted: 1, invalidCountName: "added_data_files"},
+ {name: "v2 existing data files", version: 2, added: 1,
existing: -123, deleted: 1, invalidCountName: "existing_data_files"},
Review Comment:
This table only covers data manifests — every row is the default
`ManifestContentData`. The delete-file counts go through the same
`appendManifestCount`, so a negative delete-count on a V2/V3 delete manifest
hits the identical error path but isn't verified. I'd add a few
`ManifestContentDeletes` rows with negative counts to cover the symmetry.
--
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]
