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 2025/02/21 22:11:13 UTC
(iceberg-go) 01/06: Fix #305: ManifestEntry partition field schema should be dynamically built
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch fix-manifest-entry-partition-schema
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
commit f39f69d7061487729959774290ed739f028b3534
Author: Arnaud Briche <br...@gmail.com>
AuthorDate: Mon Feb 17 16:51:51 2025 +0100
Fix #305: ManifestEntry partition field schema should be dynamically built
---
internal/avro_schemas.go | 568 ----------------------------------------
internal/manifest-entry-v1.json | 182 +++++++++++++
internal/manifest-entry-v2.json | 190 ++++++++++++++
internal/manifest-list-v1.json | 88 +++++++
internal/manifest-list-v2.json | 71 +++++
internal/manifest_entry_v1.go | 278 ++++++++++++++++++++
internal/manifest_entry_v2.go | 178 +++++++++++++
internal/manifest_list_v1.go | 142 ++++++++++
internal/manifest_list_v2.go | 100 +++++++
internal/utils.go | 35 +++
manifest.go | 48 +++-
manifest_test.go | 21 +-
utils.go | 74 +++++-
13 files changed, 1390 insertions(+), 585 deletions(-)
diff --git a/internal/avro_schemas.go b/internal/avro_schemas.go
deleted file mode 100644
index 893b7b2..0000000
--- a/internal/avro_schemas.go
+++ /dev/null
@@ -1,568 +0,0 @@
-// 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 internal
-
-import "github.com/hamba/avro/v2"
-
-const (
- ManifestListV1Key = "manifest-list-v1"
- ManifestListV2Key = "manifest-list-v2"
- ManifestEntryV1Key = "manifest-entry-v1"
- ManifestEntryV2Key = "manifest-entry-v2"
-)
-
-var (
- AvroSchemaCache avro.SchemaCache
-)
-
-func init() {
- AvroSchemaCache.Add(ManifestListV1Key, avro.MustParse(`{
- "type": "record",
- "name": "manifest_file",
- "fields": [
- {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
- {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
- {"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
- {
- "name": "added_snapshot_id",
- "type": "long",
- "doc": "Snapshot ID that added the manifest",
- "field-id": 503
- },
- {
- "name": "added_data_files_count",
- "type": ["null", "int"],
- "doc": "Added entry count",
- "field-id": 504
- },
- {
- "name": "existing_data_files_count",
- "type": ["null", "int"],
- "doc": "Existing entry count",
- "field-id": 505
- },
- {
- "name": "deleted_data_files_count",
- "type": ["null", "int"],
- "doc": "Deleted entry count",
- "field-id": 506
- },
- {
- "name": "partitions",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "r508",
- "fields": [
- {
- "name": "contains_null",
- "type": "boolean",
- "doc": "True if any file has a null partition value",
- "field-id": 509
- },
- {
- "name": "contains_nan",
- "type": ["null", "boolean"],
- "doc": "True if any file has a nan partition value",
- "field-id": 518
- },
- {
- "name": "lower_bound",
- "type": ["null", "bytes"],
- "doc": "Partition lower bound for all files",
- "field-id": 510
- },
- {
- "name": "upper_bound",
- "type": ["null", "bytes"],
- "doc": "Partition upper bound for all files",
- "field-id": 511
- }
- ]
- },
- "element-id": 508
- }
- ],
- "doc": "Summary for each partition",
- "field-id": 507
- },
- {"name": "added_rows_count", "type": ["null", "long"], "doc": "Added rows count", "field-id": 512},
- {
- "name": "existing_rows_count",
- "type": ["null", "long"],
- "doc": "Existing rows count",
- "field-id": 513
- },
- {
- "name": "deleted_rows_count",
- "type": ["null", "long"],
- "doc": "Deleted rows count",
- "field-id": 514
- }
- ]
- }`))
-
- AvroSchemaCache.Add(ManifestListV2Key, avro.MustParse(`{
- "type": "record",
- "name": "manifest_file",
- "fields": [
- {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
- {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
- {"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
- {"name": "content", "type": "int", "doc": "Contents of the manifest: 0=data, 1=deletes", "field-id": 517},
- {
- "name": "sequence_number",
- "type": "long",
- "doc": "Sequence number when the manifest was added",
- "field-id": 515
- },
- {
- "name": "min_sequence_number",
- "type": "long",
- "doc": "Lowest sequence number in the manifest",
- "field-id": 516
- },
- {"name": "added_snapshot_id", "type": "long", "doc": "Snapshot ID that added the manifest", "field-id": 503},
- {"name": "added_files_count", "type": "int", "doc": "Added entry count", "field-id": 504},
- {"name": "existing_files_count", "type": "int", "doc": "Existing entry count", "field-id": 505},
- {"name": "deleted_files_count", "type": "int", "doc": "Deleted entry count", "field-id": 506},
- {"name": "added_rows_count", "type": "long", "doc": "Added rows count", "field-id": 512},
- {"name": "existing_rows_count", "type": "long", "doc": "Existing rows count", "field-id": 513},
- {"name": "deleted_rows_count", "type": "long", "doc": "Deleted rows count", "field-id": 514},
- {
- "name": "partitions",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "r508",
- "fields": [
- {
- "name": "contains_null",
- "type": "boolean",
- "doc": "True if any file has a null partition value",
- "field-id": 509
- },
- {
- "name": "contains_nan",
- "type": ["null", "boolean"],
- "doc": "True if any file has a nan partition value",
- "field-id": 518
- },
- {
- "name": "lower_bound",
- "type": ["null", "bytes"],
- "doc": "Partition lower bound for all files",
- "field-id": 510
- },
- {
- "name": "upper_bound",
- "type": ["null", "bytes"],
- "doc": "Partition upper bound for all files",
- "field-id": 511
- }
- ]
- },
- "element-id": 508
- }
- ],
- "doc": "Summary for each partition",
- "field-id": 507
- }
- ]
- }`))
-
- AvroSchemaCache.Add(ManifestEntryV1Key, avro.MustParse(`{
- "type": "record",
- "name": "manifest_entry",
- "fields": [
- {"name": "status", "type": "int", "field-id": 0},
- {"name": "snapshot_id", "type": "long", "field-id": 1},
- {
- "name": "data_file",
- "type": {
- "type": "record",
- "name": "r2",
- "fields": [
- {"name": "file_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 100},
- {
- "name": "file_format",
- "type": "string",
- "doc": "File format name: avro, orc, or parquet",
- "field-id": 101
- },
- {
- "name": "partition",
- "type": {
- "type": "record",
- "name": "r102",
- "fields": [
- {"field-id": 1000, "name": "VendorID", "type": ["null", "int"]},
- {
- "field-id": 1001,
- "name": "tpep_pickup_datetime",
- "type": ["null", {"type": "int", "logicalType": "date"}]
- }
- ]
- },
- "field-id": 102
- },
- {"name": "record_count", "type": "long", "doc": "Number of records in the file", "field-id": 103},
- {"name": "file_size_in_bytes", "type": "long", "doc": "Total file size in bytes", "field-id": 104},
- {"name": "block_size_in_bytes", "type": "long", "field-id": 105},
- {
- "name": "column_sizes",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k117_v118",
- "fields": [
- {"name": "key", "type": "int", "field-id": 117},
- {"name": "value", "type": "long", "field-id": 118}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to total size on disk",
- "field-id": 108
- },
- {
- "name": "value_counts",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k119_v120",
- "fields": [
- {"name": "key", "type": "int", "field-id": 119},
- {"name": "value", "type": "long", "field-id": 120}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to total count, including null and NaN",
- "field-id": 109
- },
- {
- "name": "null_value_counts",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k121_v122",
- "fields": [
- {"name": "key", "type": "int", "field-id": 121},
- {"name": "value", "type": "long", "field-id": 122}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to null value count",
- "field-id": 110
- },
- {
- "name": "nan_value_counts",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k138_v139",
- "fields": [
- {"name": "key", "type": "int", "field-id": 138},
- {"name": "value", "type": "long", "field-id": 139}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to number of NaN values in the column",
- "field-id": 137
- },
- {
- "name": "lower_bounds",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k126_v127",
- "fields": [
- {"name": "key", "type": "int", "field-id": 126},
- {"name": "value", "type": "bytes", "field-id": 127}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to lower bound",
- "field-id": 125
- },
- {
- "name": "upper_bounds",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k129_v130",
- "fields": [
- {"name": "key", "type": "int", "field-id": 129},
- {"name": "value", "type": "bytes", "field-id": 130}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to upper bound",
- "field-id": 128
- },
- {
- "name": "key_metadata",
- "type": ["null", "bytes"],
- "doc": "Encryption key metadata blob",
- "field-id": 131
- },
- {
- "name": "split_offsets",
- "type": ["null", {"type": "array", "items": "long", "element-id": 133}],
- "doc": "Splittable offsets",
- "field-id": 132
- },
- {
- "name": "sort_order_id",
- "type": ["null", "int"],
- "doc": "Sort order ID",
- "field-id": 140
- }
- ]
- },
- "field-id": 2
- }
- ]
- }`))
-
- AvroSchemaCache.Add(ManifestEntryV2Key, avro.MustParse(`{
- "type": "record",
- "name": "manifest_entry",
- "fields": [
- {"name": "status", "type": "int", "field-id": 0},
- {"name": "snapshot_id", "type": ["null", "long"], "field-id": 1},
- {"name": "sequence_number", "type": ["null", "long"], "field-id": 3},
- {"name": "file_sequence_number", "type": ["null", "long"], "field-id": 4},
- {
- "name": "data_file",
- "type": {
- "type": "record",
- "name": "r2",
- "fields": [
- {"name": "content", "type": "int", "doc": "Type of content stored by the data file", "field-id": 134},
- {"name": "file_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 100},
- {
- "name": "file_format",
- "type": "string",
- "doc": "File format name: avro, orc, or parquet",
- "field-id": 101
- },
- {
- "name": "partition",
- "type": {
- "type": "record",
- "name": "r102",
- "fields": [
- {"field-id": 1000, "name": "VendorID", "type": ["null", "int"]},
- {
- "field-id": 1001,
- "name": "tpep_pickup_datetime",
- "type": ["null", {"type": "int", "logicalType": "date"}]
- }
- ]
- },
- "field-id": 102
- },
- {"name": "record_count", "type": "long", "doc": "Number of records in the file", "field-id": 103},
- {"name": "file_size_in_bytes", "type": "long", "doc": "Total file size in bytes", "field-id": 104},
- {
- "name": "column_sizes",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k117_v118",
- "fields": [
- {"name": "key", "type": "int", "field-id": 117},
- {"name": "value", "type": "long", "field-id": 118}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to total size on disk",
- "field-id": 108
- },
- {
- "name": "value_counts",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k119_v120",
- "fields": [
- {"name": "key", "type": "int", "field-id": 119},
- {"name": "value", "type": "long", "field-id": 120}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to total count, including null and NaN",
- "field-id": 109
- },
- {
- "name": "null_value_counts",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k121_v122",
- "fields": [
- {"name": "key", "type": "int", "field-id": 121},
- {"name": "value", "type": "long", "field-id": 122}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to null value count",
- "field-id": 110
- },
- {
- "name": "nan_value_counts",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k138_v139",
- "fields": [
- {"name": "key", "type": "int", "field-id": 138},
- {"name": "value", "type": "long", "field-id": 139}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to number of NaN values in the column",
- "field-id": 137
- },
- {
- "name": "lower_bounds",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k126_v127",
- "fields": [
- {"name": "key", "type": "int", "field-id": 126},
- {"name": "value", "type": "bytes", "field-id": 127}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to lower bound",
- "field-id": 125
- },
- {
- "name": "upper_bounds",
- "type": [
- "null",
- {
- "type": "array",
- "items": {
- "type": "record",
- "name": "k129_v130",
- "fields": [
- {"name": "key", "type": "int", "field-id": 129},
- {"name": "value", "type": "bytes", "field-id": 130}
- ]
- },
- "logicalType": "map"
- }
- ],
- "doc": "Map of column id to upper bound",
- "field-id": 128
- },
- {
- "name": "key_metadata",
- "type": ["null", "bytes"],
- "doc": "Encryption key metadata blob",
- "field-id": 131
- },
- {
- "name": "split_offsets",
- "type": ["null", {"type": "array", "items": "long", "element-id": 133}],
- "doc": "Splittable offsets",
- "field-id": 132
- },
- {
- "name": "equality_ids",
- "type": ["null", {"type": "array", "items": "int", "element-id": 136}],
- "doc": "Field ids used to determine row equality for delete files",
- "field-id": 135
- },
- {
- "name": "sort_order_id",
- "type": ["null", "int"],
- "doc": "Sort order ID",
- "field-id": 140
- }
- ]
- },
- "field-id": 2
- }
- ]
- }`))
-}
diff --git a/internal/manifest-entry-v1.json b/internal/manifest-entry-v1.json
new file mode 100644
index 0000000..3430eac
--- /dev/null
+++ b/internal/manifest-entry-v1.json
@@ -0,0 +1,182 @@
+{
+ "type": "record",
+ "name": "manifest_entry",
+ "fields": [
+ {"name": "status", "type": "int", "field-id": 0},
+ {"name": "snapshot_id", "type": "long", "field-id": 1},
+ {
+ "name": "data_file",
+ "type": {
+ "type": "record",
+ "name": "r2",
+ "fields": [
+ {"name": "file_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 100},
+ {
+ "name": "file_format",
+ "type": "string",
+ "doc": "File format name: avro, orc, or parquet",
+ "field-id": 101
+ },
+ {
+ "name": "partition",
+ "type": {
+ "type": "record",
+ "name": "r102",
+ "fields": [
+ {"field-id": 1000, "name": "VendorID", "type": ["null", "int"]},
+ {
+ "field-id": 1001,
+ "name": "tpep_pickup_datetime",
+ "type": ["null", {"type": "int", "logicalType": "date"}]
+ }
+ ]
+ },
+ "field-id": 102
+ },
+ {"name": "record_count", "type": "long", "doc": "Number of records in the file", "field-id": 103},
+ {"name": "file_size_in_bytes", "type": "long", "doc": "Total file size in bytes", "field-id": 104},
+ {"name": "block_size_in_bytes", "type": "long", "field-id": 105},
+ {
+ "name": "column_sizes",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k117_v118",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 117},
+ {"name": "value", "type": "long", "field-id": 118}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to total size on disk",
+ "field-id": 108
+ },
+ {
+ "name": "value_counts",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k119_v120",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 119},
+ {"name": "value", "type": "long", "field-id": 120}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to total count, including null and NaN",
+ "field-id": 109
+ },
+ {
+ "name": "null_value_counts",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k121_v122",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 121},
+ {"name": "value", "type": "long", "field-id": 122}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to null value count",
+ "field-id": 110
+ },
+ {
+ "name": "nan_value_counts",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k138_v139",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 138},
+ {"name": "value", "type": "long", "field-id": 139}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to number of NaN values in the column",
+ "field-id": 137
+ },
+ {
+ "name": "lower_bounds",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k126_v127",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 126},
+ {"name": "value", "type": "bytes", "field-id": 127}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to lower bound",
+ "field-id": 125
+ },
+ {
+ "name": "upper_bounds",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k129_v130",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 129},
+ {"name": "value", "type": "bytes", "field-id": 130}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to upper bound",
+ "field-id": 128
+ },
+ {
+ "name": "key_metadata",
+ "type": ["null", "bytes"],
+ "doc": "Encryption key metadata blob",
+ "field-id": 131
+ },
+ {
+ "name": "split_offsets",
+ "type": ["null", {"type": "array", "items": "long", "element-id": 133}],
+ "doc": "Splittable offsets",
+ "field-id": 132
+ },
+ {
+ "name": "sort_order_id",
+ "type": ["null", "int"],
+ "doc": "Sort order ID",
+ "field-id": 140
+ }
+ ]
+ },
+ "field-id": 2
+ }
+ ]
+}
\ No newline at end of file
diff --git a/internal/manifest-entry-v2.json b/internal/manifest-entry-v2.json
new file mode 100644
index 0000000..9bf073a
--- /dev/null
+++ b/internal/manifest-entry-v2.json
@@ -0,0 +1,190 @@
+{
+ "type": "record",
+ "name": "manifest_entry",
+ "fields": [
+ {"name": "status", "type": "int", "field-id": 0},
+ {"name": "snapshot_id", "type": ["null", "long"], "field-id": 1},
+ {"name": "sequence_number", "type": ["null", "long"], "field-id": 3},
+ {"name": "file_sequence_number", "type": ["null", "long"], "field-id": 4},
+ {
+ "name": "data_file",
+ "type": {
+ "type": "record",
+ "name": "r2",
+ "fields": [
+ {"name": "content", "type": "int", "doc": "Type of content stored by the data file", "field-id": 134},
+ {"name": "file_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 100},
+ {
+ "name": "file_format",
+ "type": "string",
+ "doc": "File format name: avro, orc, or parquet",
+ "field-id": 101
+ },
+ {
+ "name": "partition",
+ "type": {
+ "type": "record",
+ "name": "r102",
+ "fields": [
+ {"field-id": 1000, "name": "VendorID", "type": ["null", "int"]},
+ {
+ "field-id": 1001,
+ "name": "tpep_pickup_datetime",
+ "type": ["null", {"type": "int", "logicalType": "date"}]
+ }
+ ]
+ },
+ "field-id": 102
+ },
+ {"name": "record_count", "type": "long", "doc": "Number of records in the file", "field-id": 103},
+ {"name": "file_size_in_bytes", "type": "long", "doc": "Total file size in bytes", "field-id": 104},
+ {
+ "name": "column_sizes",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k117_v118",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 117},
+ {"name": "value", "type": "long", "field-id": 118}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to total size on disk",
+ "field-id": 108
+ },
+ {
+ "name": "value_counts",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k119_v120",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 119},
+ {"name": "value", "type": "long", "field-id": 120}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to total count, including null and NaN",
+ "field-id": 109
+ },
+ {
+ "name": "null_value_counts",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k121_v122",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 121},
+ {"name": "value", "type": "long", "field-id": 122}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to null value count",
+ "field-id": 110
+ },
+ {
+ "name": "nan_value_counts",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k138_v139",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 138},
+ {"name": "value", "type": "long", "field-id": 139}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to number of NaN values in the column",
+ "field-id": 137
+ },
+ {
+ "name": "lower_bounds",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k126_v127",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 126},
+ {"name": "value", "type": "bytes", "field-id": 127}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to lower bound",
+ "field-id": 125
+ },
+ {
+ "name": "upper_bounds",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "k129_v130",
+ "fields": [
+ {"name": "key", "type": "int", "field-id": 129},
+ {"name": "value", "type": "bytes", "field-id": 130}
+ ]
+ },
+ "logicalType": "map"
+ }
+ ],
+ "doc": "Map of column id to upper bound",
+ "field-id": 128
+ },
+ {
+ "name": "key_metadata",
+ "type": ["null", "bytes"],
+ "doc": "Encryption key metadata blob",
+ "field-id": 131
+ },
+ {
+ "name": "split_offsets",
+ "type": ["null", {"type": "array", "items": "long", "element-id": 133}],
+ "doc": "Splittable offsets",
+ "field-id": 132
+ },
+ {
+ "name": "equality_ids",
+ "type": ["null", {"type": "array", "items": "int", "element-id": 136}],
+ "doc": "Field ids used to determine row equality for delete files",
+ "field-id": 135
+ },
+ {
+ "name": "sort_order_id",
+ "type": ["null", "int"],
+ "doc": "Sort order ID",
+ "field-id": 140
+ }
+ ]
+ },
+ "field-id": 2
+ }
+ ]
+}
\ No newline at end of file
diff --git a/internal/manifest-list-v1.json b/internal/manifest-list-v1.json
new file mode 100644
index 0000000..aee7429
--- /dev/null
+++ b/internal/manifest-list-v1.json
@@ -0,0 +1,88 @@
+{
+ "type": "record",
+ "name": "manifest_file",
+ "fields": [
+ {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
+ {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
+ {"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
+ {
+ "name": "added_snapshot_id",
+ "type": "long",
+ "doc": "Snapshot ID that added the manifest",
+ "field-id": 503
+ },
+ {
+ "name": "added_data_files_count",
+ "type": ["null", "int"],
+ "doc": "Added entry count",
+ "field-id": 504
+ },
+ {
+ "name": "existing_data_files_count",
+ "type": ["null", "int"],
+ "doc": "Existing entry count",
+ "field-id": 505
+ },
+ {
+ "name": "deleted_data_files_count",
+ "type": ["null", "int"],
+ "doc": "Deleted entry count",
+ "field-id": 506
+ },
+ {
+ "name": "partitions",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "r508",
+ "fields": [
+ {
+ "name": "contains_null",
+ "type": "boolean",
+ "doc": "True if any file has a null partition value",
+ "field-id": 509
+ },
+ {
+ "name": "contains_nan",
+ "type": ["null", "boolean"],
+ "doc": "True if any file has a nan partition value",
+ "field-id": 518
+ },
+ {
+ "name": "lower_bound",
+ "type": ["null", "bytes"],
+ "doc": "Partition lower bound for all files",
+ "field-id": 510
+ },
+ {
+ "name": "upper_bound",
+ "type": ["null", "bytes"],
+ "doc": "Partition upper bound for all files",
+ "field-id": 511
+ }
+ ]
+ },
+ "element-id": 508
+ }
+ ],
+ "doc": "Summary for each partition",
+ "field-id": 507
+ },
+ {"name": "added_rows_count", "type": ["null", "long"], "doc": "Added rows count", "field-id": 512},
+ {
+ "name": "existing_rows_count",
+ "type": ["null", "long"],
+ "doc": "Existing rows count",
+ "field-id": 513
+ },
+ {
+ "name": "deleted_rows_count",
+ "type": ["null", "long"],
+ "doc": "Deleted rows count",
+ "field-id": 514
+ }
+ ]
+}
\ No newline at end of file
diff --git a/internal/manifest-list-v2.json b/internal/manifest-list-v2.json
new file mode 100644
index 0000000..0f933a0
--- /dev/null
+++ b/internal/manifest-list-v2.json
@@ -0,0 +1,71 @@
+{
+ "type": "record",
+ "name": "manifest_file",
+ "fields": [
+ {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
+ {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
+ {"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
+ {"name": "content", "type": "int", "doc": "Contents of the manifest: 0=data, 1=deletes", "field-id": 517},
+ {
+ "name": "sequence_number",
+ "type": "long",
+ "doc": "Sequence number when the manifest was added",
+ "field-id": 515
+ },
+ {
+ "name": "min_sequence_number",
+ "type": "long",
+ "doc": "Lowest sequence number in the manifest",
+ "field-id": 516
+ },
+ {"name": "added_snapshot_id", "type": "long", "doc": "Snapshot ID that added the manifest", "field-id": 503},
+ {"name": "added_files_count", "type": "int", "doc": "Added entry count", "field-id": 504},
+ {"name": "existing_files_count", "type": "int", "doc": "Existing entry count", "field-id": 505},
+ {"name": "deleted_files_count", "type": "int", "doc": "Deleted entry count", "field-id": 506},
+ {"name": "added_rows_count", "type": "long", "doc": "Added rows count", "field-id": 512},
+ {"name": "existing_rows_count", "type": "long", "doc": "Existing rows count", "field-id": 513},
+ {"name": "deleted_rows_count", "type": "long", "doc": "Deleted rows count", "field-id": 514},
+ {
+ "name": "partitions",
+ "type": [
+ "null",
+ {
+ "type": "array",
+ "items": {
+ "type": "record",
+ "name": "r508",
+ "fields": [
+ {
+ "name": "contains_null",
+ "type": "boolean",
+ "doc": "True if any file has a null partition value",
+ "field-id": 509
+ },
+ {
+ "name": "contains_nan",
+ "type": ["null", "boolean"],
+ "doc": "True if any file has a nan partition value",
+ "field-id": 518
+ },
+ {
+ "name": "lower_bound",
+ "type": ["null", "bytes"],
+ "doc": "Partition lower bound for all files",
+ "field-id": 510
+ },
+ {
+ "name": "upper_bound",
+ "type": ["null", "bytes"],
+ "doc": "Partition upper bound for all files",
+ "field-id": 511
+ }
+ ]
+ },
+ "element-id": 508
+ }
+ ],
+ "doc": "Summary for each partition",
+ "field-id": 507
+ }
+ ]
+}
\ No newline at end of file
diff --git a/internal/manifest_entry_v1.go b/internal/manifest_entry_v1.go
new file mode 100644
index 0000000..225ce17
--- /dev/null
+++ b/internal/manifest_entry_v1.go
@@ -0,0 +1,278 @@
+package internal
+
+import "github.com/hamba/avro/v2"
+
+func MustNewManifestEntryV1Schema(partitionSchema avro.Schema) avro.Schema {
+ return MustNewRecordSchema("manifest_entry", "", []*avro.Field{
+ MustNewField(
+ "status",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 0}),
+ ),
+ MustNewField(
+ "snapshot_id",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"field-id": 1}),
+ ),
+ MustNewField(
+ "data_file",
+ MustNewDataFileV1Schema(partitionSchema),
+ avro.WithProps(map[string]any{"field-id": 2}),
+ ),
+ })
+}
+
+func MustNewDataFileV1Schema(partitionSchema avro.Schema) avro.Schema {
+ return MustNewRecordSchema("r2", "", []*avro.Field{
+ MustNewField(
+ "file_path",
+ avro.NewPrimitiveSchema(avro.String, nil),
+ avro.WithDoc("Location URI with FS scheme"),
+ avro.WithProps(map[string]any{"field-id": 100}),
+ ),
+ MustNewField(
+ "file_format",
+ avro.NewPrimitiveSchema(avro.String, nil),
+ avro.WithDoc("File format name: avro, orc, or parquet"),
+ avro.WithProps(map[string]any{"field-id": 101}),
+ ),
+ MustNewField(
+ "partition",
+ partitionSchema,
+ avro.WithProps(map[string]any{"field-id": 102}),
+ ),
+ MustNewField(
+ "record_count",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Number of records in the file"),
+ avro.WithProps(map[string]any{"field-id": 103}),
+ ),
+ MustNewField(
+ "file_size_in_bytes",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Total file size in bytes"),
+ avro.WithProps(map[string]any{"field-id": 104}),
+ ),
+ MustNewField(
+ "block_size_in_bytes",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"field-id": 105}),
+ ),
+ MustNewField(
+ "column_sizes",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewColumnSizesMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to total size on disk"),
+ avro.WithProps(map[string]any{"field-id": 108}),
+ ),
+ MustNewField(
+ "value_counts",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewValueCountsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to total count, including null and NaN"),
+ avro.WithProps(map[string]any{"field-id": 109}),
+ ),
+ MustNewField(
+ "null_value_counts",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewNullValueCountsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to null value count"),
+ avro.WithProps(map[string]any{"field-id": 110}),
+ ),
+ MustNewField(
+ "nan_value_counts",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewNanValueCountsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to number of NaN values in the column"),
+ avro.WithProps(map[string]any{"field-id": 137}),
+ ),
+ MustNewField(
+ "lower_bounds",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewLowerBoundsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to lower bound"),
+ avro.WithProps(map[string]any{"field-id": 125}),
+ ),
+ MustNewField(
+ "upper_bounds",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewUpperBoundsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to upper bound"),
+ avro.WithProps(map[string]any{"field-id": 128}),
+ ),
+ MustNewField(
+ "key_metadata",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Bytes, nil),
+ }),
+ avro.WithDoc("Encryption key metadata blob"),
+ avro.WithProps(map[string]any{"field-id": 131}),
+ ),
+ MustNewField(
+ "split_offsets",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewArraySchema(
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"element-id": 133}),
+ ),
+ }),
+ avro.WithDoc("Splittable offsets"),
+ avro.WithProps(map[string]any{"field-id": 132}),
+ ),
+ MustNewField(
+ "sort_order_id",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ }),
+ avro.WithDoc("Sort order ID"),
+ avro.WithProps(map[string]any{"field-id": 140}),
+ ),
+ })
+}
+
+func MustNewColumnSizesMapSchema() avro.Schema {
+ return avro.NewArraySchema(
+ MustNewRecordSchema(
+ "k117_v118",
+ "",
+ []*avro.Field{
+ MustNewField(
+ "key",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 117}),
+ ),
+ MustNewField(
+ "value",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"field-id": 118}),
+ ),
+ },
+ ),
+ avro.WithProps(map[string]any{"logicalType": "map"}),
+ )
+}
+
+func MustNewValueCountsMapSchema() avro.Schema {
+ return avro.NewArraySchema(
+ MustNewRecordSchema(
+ "k119_v120",
+ "",
+ []*avro.Field{
+ MustNewField(
+ "key",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 119}),
+ ),
+ MustNewField(
+ "value",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"field-id": 120}),
+ ),
+ },
+ ),
+ avro.WithProps(map[string]any{"logicalType": "map"}),
+ )
+}
+
+func MustNewNullValueCountsMapSchema() avro.Schema {
+ return avro.NewArraySchema(
+ MustNewRecordSchema(
+ "k121_v122",
+ "",
+ []*avro.Field{
+ MustNewField(
+ "key",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 121}),
+ ),
+ MustNewField(
+ "value",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"field-id": 122}),
+ ),
+ },
+ ),
+ avro.WithProps(map[string]any{"logicalType": "map"}),
+ )
+}
+
+func MustNewNanValueCountsMapSchema() avro.Schema {
+ return avro.NewArraySchema(
+ MustNewRecordSchema(
+ "k138_v139",
+ "",
+ []*avro.Field{
+ MustNewField(
+ "key",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 138}),
+ ),
+ MustNewField(
+ "value",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"field-id": 139}),
+ ),
+ },
+ ),
+ avro.WithProps(map[string]any{"logicalType": "map"}),
+ )
+}
+
+func MustNewLowerBoundsMapSchema() avro.Schema {
+ return avro.NewArraySchema(
+ MustNewRecordSchema(
+ "k126_v127",
+ "",
+ []*avro.Field{
+ MustNewField(
+ "key",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 126}),
+ ),
+ MustNewField(
+ "value",
+ avro.NewPrimitiveSchema(avro.Bytes, nil),
+ avro.WithProps(map[string]any{"field-id": 127}),
+ ),
+ },
+ ),
+ avro.WithProps(map[string]any{"logicalType": "map"}),
+ )
+}
+
+func MustNewUpperBoundsMapSchema() avro.Schema {
+ return avro.NewArraySchema(
+ MustNewRecordSchema(
+ "k129_v130",
+ "",
+ []*avro.Field{
+ MustNewField(
+ "key",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 129}),
+ ),
+ MustNewField(
+ "value",
+ avro.NewPrimitiveSchema(avro.Bytes, nil),
+ avro.WithProps(map[string]any{"field-id": 130}),
+ ),
+ },
+ ),
+ avro.WithProps(map[string]any{"logicalType": "map"}),
+ )
+}
diff --git a/internal/manifest_entry_v2.go b/internal/manifest_entry_v2.go
new file mode 100644
index 0000000..ae6f6ab
--- /dev/null
+++ b/internal/manifest_entry_v2.go
@@ -0,0 +1,178 @@
+package internal
+
+import "github.com/hamba/avro/v2"
+
+func MustNewManifestEntryV2Schema(partitionSchema avro.Schema) avro.Schema {
+ return MustNewRecordSchema("manifest_entry", "", []*avro.Field{
+ MustNewField(
+ "status",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"field-id": 0}),
+ ),
+ MustNewField(
+ "snapshot_id",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ }),
+ avro.WithProps(map[string]any{"field-id": 1}),
+ ),
+ MustNewField(
+ "sequence_number",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ }),
+ avro.WithProps(map[string]any{"field-id": 3}),
+ ),
+ MustNewField(
+ "file_sequence_number",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ }),
+ avro.WithProps(map[string]any{"field-id": 4}),
+ ),
+ MustNewField(
+ "data_file",
+ MustNewDataFileV2Schema(partitionSchema),
+ avro.WithProps(map[string]any{"field-id": 2}),
+ ),
+ })
+}
+
+func MustNewDataFileV2Schema(partitionSchema avro.Schema) avro.Schema {
+ return MustNewRecordSchema("r2", "", []*avro.Field{
+ MustNewField(
+ "content",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithDoc("Type of content stored by the data file"),
+ avro.WithProps(map[string]any{"field-id": 134}),
+ ),
+ MustNewField(
+ "file_path",
+ avro.NewPrimitiveSchema(avro.String, nil),
+ avro.WithDoc("Location URI with FS scheme"),
+ avro.WithProps(map[string]any{"field-id": 100}),
+ ),
+ MustNewField(
+ "file_format",
+ avro.NewPrimitiveSchema(avro.String, nil),
+ avro.WithDoc("File format name: avro, orc, or parquet"),
+ avro.WithProps(map[string]any{"field-id": 101}),
+ ),
+ MustNewField(
+ "partition",
+ partitionSchema,
+ avro.WithProps(map[string]any{"field-id": 102}),
+ ),
+ MustNewField(
+ "record_count",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Number of records in the file"),
+ avro.WithProps(map[string]any{"field-id": 103}),
+ ),
+ MustNewField(
+ "file_size_in_bytes",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Total file size in bytes"),
+ avro.WithProps(map[string]any{"field-id": 104}),
+ ),
+ MustNewField(
+ "column_sizes",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewColumnSizesMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to total size on disk"),
+ avro.WithProps(map[string]any{"field-id": 108}),
+ ),
+ MustNewField(
+ "value_counts",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewValueCountsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to total count, including null and NaN"),
+ avro.WithProps(map[string]any{"field-id": 109}),
+ ),
+ MustNewField(
+ "null_value_counts",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewNullValueCountsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to null value count"),
+ avro.WithProps(map[string]any{"field-id": 110}),
+ ),
+ MustNewField(
+ "nan_value_counts",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewNanValueCountsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to number of NaN values in the column"),
+ avro.WithProps(map[string]any{"field-id": 137}),
+ ),
+ MustNewField(
+ "lower_bounds",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewLowerBoundsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to lower bound"),
+ avro.WithProps(map[string]any{"field-id": 125}),
+ ),
+ MustNewField(
+ "upper_bounds",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ MustNewUpperBoundsMapSchema(),
+ }),
+ avro.WithDoc("Map of column id to upper bound"),
+ avro.WithProps(map[string]any{"field-id": 128}),
+ ),
+ MustNewField(
+ "key_metadata",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Bytes, nil),
+ }),
+ avro.WithDoc("Encryption key metadata blob"),
+ avro.WithProps(map[string]any{"field-id": 131}),
+ ),
+ MustNewField(
+ "split_offsets",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewArraySchema(
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithProps(map[string]any{"element-id": 133}),
+ ),
+ }),
+ avro.WithDoc("Splittable offsets"),
+ avro.WithProps(map[string]any{"field-id": 132}),
+ ),
+ MustNewField(
+ "equality_ids",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewArraySchema(
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithProps(map[string]any{"element-id": 136}),
+ ),
+ }),
+ avro.WithDoc("Field ids used to determine row equality for delete files"),
+ avro.WithProps(map[string]any{"field-id": 135}),
+ ),
+ MustNewField(
+ "sort_order_id",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ }),
+ avro.WithDoc("Sort order ID"),
+ avro.WithProps(map[string]any{"field-id": 140}),
+ ),
+ })
+}
diff --git a/internal/manifest_list_v1.go b/internal/manifest_list_v1.go
new file mode 100644
index 0000000..92b5fd5
--- /dev/null
+++ b/internal/manifest_list_v1.go
@@ -0,0 +1,142 @@
+package internal
+
+import (
+ "github.com/hamba/avro/v2"
+)
+
+func MustNewManifestListV1Schema() avro.Schema {
+ return MustNewRecordSchema("manifest_file", "", []*avro.Field{
+ MustNewField(
+ "manifest_path",
+ avro.NewPrimitiveSchema(avro.String, nil),
+ avro.WithDoc("Location URI with FS scheme"),
+ avro.WithProps(map[string]any{"field-id": 500}),
+ ),
+ MustNewField(
+ "manifest_length",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Total file size in bytes"),
+ avro.WithProps(map[string]any{"field-id": 501}),
+ ),
+ MustNewField(
+ "partition_spec_id",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithDoc("Spec ID used to write"),
+ avro.WithProps(map[string]any{"field-id": 502}),
+ ),
+ MustNewField(
+ "added_snapshot_id",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Snapshot ID that added the manifest"),
+ avro.WithProps(map[string]any{"field-id": 503}),
+ ),
+ MustNewField(
+ "added_data_files_count",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ }),
+ avro.WithDoc("Added entry count"),
+ avro.WithProps(map[string]any{"field-id": 504}),
+ ),
+ MustNewField(
+ "existing_data_files_count",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ }),
+ avro.WithDoc("Existing entry count"),
+ avro.WithProps(map[string]any{"field-id": 505}),
+ ),
+ MustNewField(
+ "deleted_data_files_count",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ }),
+ avro.WithDoc("Deleted entry count"),
+ avro.WithProps(map[string]any{"field-id": 506}),
+ ),
+ MustNewField(
+ "partitions",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewArraySchema(
+ MustNewPartitionSchema(),
+ avro.WithProps(map[string]any{"element-id": 508}),
+ ),
+ }),
+ avro.WithDoc("Summary for each partition"),
+ avro.WithProps(map[string]any{"field-id": 507}),
+ ),
+ MustNewField(
+ "added_rows_count",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ }),
+ avro.WithDoc("Added rows count"),
+ avro.WithProps(map[string]any{"field-id": 512}),
+ ),
+ MustNewField(
+ "existing_rows_count",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ }),
+ avro.WithDoc("Existing rows count"),
+ avro.WithProps(map[string]any{"field-id": 513}),
+ ),
+ MustNewField(
+ "deleted_rows_count",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ }),
+ avro.WithDoc("Deleted rows count"),
+ avro.WithProps(map[string]any{"field-id": 514}),
+ ),
+ })
+}
+
+func MustNewPartitionSchema() avro.Schema {
+ return MustNewRecordSchema(
+ "r508",
+ "",
+ []*avro.Field{
+ MustNewField(
+ "contains_null",
+ avro.NewPrimitiveSchema(avro.Boolean, nil),
+ avro.WithDoc("True if any file has a null partition value"),
+ avro.WithProps(map[string]any{"field-id": 509}),
+ ),
+ MustNewField(
+ "contains_nan",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Boolean, nil),
+ }),
+ avro.WithDoc("True if any file has a nan partition value"),
+ avro.WithProps(map[string]any{"field-id": 518}),
+ ),
+ MustNewField(
+ "lower_bound",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Bytes, nil),
+ }),
+ avro.WithDoc("Partition lower bound for all files"),
+ avro.WithProps(map[string]any{"field-id": 510}),
+ ),
+ MustNewField(
+ "upper_bound",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewPrimitiveSchema(avro.Bytes, nil),
+ }),
+ avro.WithDoc("Partition upper bound for all files"),
+ avro.WithProps(map[string]any{"field-id": 511}),
+ ),
+ },
+ )
+}
diff --git a/internal/manifest_list_v2.go b/internal/manifest_list_v2.go
new file mode 100644
index 0000000..77052fb
--- /dev/null
+++ b/internal/manifest_list_v2.go
@@ -0,0 +1,100 @@
+package internal
+
+import (
+ "github.com/hamba/avro/v2"
+)
+
+func MustNewManifestListV2Schema() avro.Schema {
+ return MustNewRecordSchema("manifest_file", "", []*avro.Field{
+ MustNewField(
+ "manifest_path",
+ avro.NewPrimitiveSchema(avro.String, nil),
+ avro.WithDoc("Location URI with FS scheme"),
+ avro.WithProps(map[string]any{"field-id": 500}),
+ ),
+ MustNewField(
+ "manifest_length",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Total file size in bytes"),
+ avro.WithProps(map[string]any{"field-id": 501}),
+ ),
+ MustNewField(
+ "partition_spec_id",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithDoc("Spec ID used to write"),
+ avro.WithProps(map[string]any{"field-id": 502}),
+ ),
+ MustNewField(
+ "content",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithDoc("Contents of the manifest: 0=data, 1=deletes"),
+ avro.WithProps(map[string]any{"field-id": 517}),
+ ),
+ MustNewField(
+ "sequence_number",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Sequence number when the manifest was added"),
+ avro.WithProps(map[string]any{"field-id": 515}),
+ ),
+ MustNewField(
+ "min_sequence_number",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Lowest sequence number in the manifest"),
+ avro.WithProps(map[string]any{"field-id": 516}),
+ ),
+ MustNewField(
+ "added_snapshot_id",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Snapshot ID that added the manifest"),
+ avro.WithProps(map[string]any{"field-id": 503}),
+ ),
+ MustNewField(
+ "added_files_count",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithDoc("Added entry count"),
+ avro.WithProps(map[string]any{"field-id": 504}),
+ ),
+ MustNewField(
+ "existing_files_count",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithDoc("Existing entry count"),
+ avro.WithProps(map[string]any{"field-id": 505}),
+ ),
+ MustNewField(
+ "deleted_files_count",
+ avro.NewPrimitiveSchema(avro.Int, nil),
+ avro.WithDoc("Deleted entry count"),
+ avro.WithProps(map[string]any{"field-id": 506}),
+ ),
+ MustNewField(
+ "partitions",
+ MustNewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ avro.NewArraySchema(
+ MustNewPartitionSchema(),
+ avro.WithProps(map[string]any{"element-id": 508}),
+ ),
+ }),
+ avro.WithDoc("Summary for each partition"),
+ avro.WithProps(map[string]any{"field-id": 507}),
+ ),
+ MustNewField(
+ "added_rows_count",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Added rows count"),
+ avro.WithProps(map[string]any{"field-id": 512}),
+ ),
+ MustNewField(
+ "existing_rows_count",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Existing rows count"),
+ avro.WithProps(map[string]any{"field-id": 513}),
+ ),
+ MustNewField(
+ "deleted_rows_count",
+ avro.NewPrimitiveSchema(avro.Long, nil),
+ avro.WithDoc("Deleted rows count"),
+ avro.WithProps(map[string]any{"field-id": 514}),
+ ),
+ })
+}
diff --git a/internal/utils.go b/internal/utils.go
new file mode 100644
index 0000000..97512de
--- /dev/null
+++ b/internal/utils.go
@@ -0,0 +1,35 @@
+package internal
+
+import (
+ "github.com/hamba/avro/v2"
+)
+
+func MustNewUnionSchema(types []avro.Schema, opts ...avro.SchemaOption) *avro.UnionSchema {
+ s, err := avro.NewUnionSchema(types, opts...)
+
+ if err != nil {
+ panic(err)
+ }
+
+ return s
+}
+
+func MustNewRecordSchema(name, namespace string, fields []*avro.Field, opts ...avro.SchemaOption) *avro.RecordSchema {
+ s, err := avro.NewRecordSchema(name, namespace, fields, opts...)
+
+ if err != nil {
+ panic(err)
+ }
+
+ return s
+}
+
+func MustNewField(name string, typ avro.Schema, opts ...avro.SchemaOption) *avro.Field {
+ f, err := avro.NewField(name, typ, opts...)
+
+ if err != nil {
+ panic(err)
+ }
+
+ return f
+}
diff --git a/manifest.go b/manifest.go
index 208e09a..b9867cb 100644
--- a/manifest.go
+++ b/manifest.go
@@ -67,6 +67,11 @@ func NewManifestV1Builder(path string, length int64, partitionSpecID int32, adde
}
}
+func (b *ManifestV1Builder) PartitionType(st *StructType) *ManifestV1Builder {
+ b.m.partitionType = st
+ return b
+}
+
func (b *ManifestV1Builder) AddedFiles(cnt int32) *ManifestV1Builder {
b.m.AddedFilesCount = &cnt
return b
@@ -112,6 +117,10 @@ func (b *ManifestV1Builder) KeyMetadata(km []byte) *ManifestV1Builder {
// a pointer to the constructed manifest file. Further calls to the modifier
// methods after calling build would modify the constructed ManifestFile.
func (b *ManifestV1Builder) Build() ManifestFile {
+ if b.m.partitionType == nil {
+ b.m.partitionType = UnpartitionedSpec.PartitionType(nil)
+ }
+
return b.m
}
@@ -126,6 +135,7 @@ func (f *fallbackManifestFileV1) toManifest() *manifestFileV1 {
}
type manifestFileV1 struct {
+ partitionType *StructType
Path string `avro:"manifest_path"`
Len int64 `avro:"manifest_length"`
SpecID int32 `avro:"partition_spec_id"`
@@ -217,7 +227,7 @@ func (m *manifestFileV1) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
// WriteEntries writes a list of manifest entries to an avro file.
func (m *manifestFileV1) WriteEntries(out io.Writer, entries []ManifestEntry) error {
- return writeManifestEntries(out, entries, m.Version())
+ return writeManifestEntries(out, m.partitionType, entries, m.Version())
}
// ManifestV2Builder is a helper for building a V2 manifest file
@@ -242,6 +252,11 @@ func NewManifestV2Builder(path string, length int64, partitionSpecID int32, cont
}
}
+func (b *ManifestV2Builder) PartitionSchema(st *StructType) *ManifestV2Builder {
+ b.m.partitionType = st
+ return b
+}
+
func (b *ManifestV2Builder) SequenceNum(num, minSeqNum int64) *ManifestV2Builder {
b.m.SeqNumber, b.m.MinSeqNumber = num, minSeqNum
return b
@@ -292,10 +307,15 @@ func (b *ManifestV2Builder) KeyMetadata(km []byte) *ManifestV2Builder {
// a pointer to the constructed manifest file. Further calls to the modifier
// methods after calling build would modify the constructed ManifestFile.
func (b *ManifestV2Builder) Build() ManifestFile {
+ if b.m.partitionType == nil {
+ b.m.partitionType = UnpartitionedSpec.PartitionType(nil)
+ }
+
return b.m
}
type manifestFileV2 struct {
+ partitionType *StructType
Path string `avro:"manifest_path"`
Len int64 `avro:"manifest_length"`
SpecID int32 `avro:"partition_spec_id"`
@@ -372,7 +392,7 @@ func (m *manifestFileV2) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
// WriteEntries writes a list of manifest entries to an avro file.
func (m *manifestFileV2) WriteEntries(out io.Writer, entries []ManifestEntry) error {
- return writeManifestEntries(out, entries, m.Version())
+ return writeManifestEntries(out, m.partitionType, entries, m.Version())
}
func getFieldIDMap(sc avro.Schema) (map[string]int, map[int]avro.LogicalType) {
@@ -613,31 +633,37 @@ func WriteManifestList(out io.Writer, files []ManifestFile) error {
}
}
- var key string
+ var sch avro.Schema
switch version {
case 1:
- key = internal.ManifestListV1Key
+ sch = internal.MustNewManifestListV1Schema()
case 2:
- key = internal.ManifestListV2Key
+ sch = internal.MustNewManifestListV2Schema()
default:
return fmt.Errorf("%w: non-recognized version %d", ErrInvalidArgument, version)
}
- return avroEncode(key, version, files, out)
+ return avroEncode(sch, version, files, out)
}
-func writeManifestEntries(out io.Writer, entries []ManifestEntry, version int) error {
- var key string
+func writeManifestEntries(out io.Writer, partitionType *StructType, entries []ManifestEntry, version int) error {
+ partitionSchema, err := structTypeToAvroPartitionSchema(partitionType)
+
+ if err != nil {
+ return err
+ }
+
+ var sch avro.Schema
switch version {
case 1:
- key = internal.ManifestEntryV1Key
+ sch = internal.MustNewManifestEntryV1Schema(partitionSchema)
case 2:
- key = internal.ManifestEntryV2Key
+ sch = internal.MustNewManifestEntryV2Schema(partitionSchema)
default:
return fmt.Errorf("%w: non-recognized version %d", ErrInvalidArgument, version)
}
- return avroEncode(key, version, entries, out)
+ return avroEncode(sch, version, entries, out)
}
// ManifestEntryStatus defines constants for the entry status of
diff --git a/manifest_test.go b/manifest_test.go
index 0e6e04c..99a0fd6 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -393,8 +393,25 @@ func (m *ManifestTestSuite) writeManifestEntries() {
manifestEntryV2Recs[i] = rec
}
- m.Require().NoError(writeManifestEntries(&m.v1ManifestEntries, manifestEntryV1Recs, 1))
- m.Require().NoError(writeManifestEntries(&m.v2ManifestEntries, manifestEntryV2Recs, 2))
+ var partitionType = StructType{
+ FieldList: []NestedField{
+ {
+ ID: 1000,
+ Name: "VendorID",
+ Type: &Int32Type{},
+ Required: false,
+ },
+ {
+ ID: 1001,
+ Name: "tpep_pickup_datetime",
+ Type: &DateType{},
+ Required: false,
+ },
+ },
+ }
+
+ m.Require().NoError(writeManifestEntries(&m.v1ManifestEntries, &partitionType, manifestEntryV1Recs, 1))
+ m.Require().NoError(writeManifestEntries(&m.v2ManifestEntries, &partitionType, manifestEntryV2Recs, 2))
}
func (m *ManifestTestSuite) SetupSuite() {
diff --git a/utils.go b/utils.go
index 087fb25..597babe 100644
--- a/utils.go
+++ b/utils.go
@@ -27,7 +27,7 @@ import (
"strconv"
"strings"
- "github.com/apache/iceberg-go/internal"
+ "github.com/hamba/avro/v2"
"github.com/hamba/avro/v2/ocf"
)
@@ -218,10 +218,11 @@ func Difference(a, b []string) []string {
return diff
}
-func avroEncode[T any](key string, version int, vals []T, out io.Writer) error {
+func avroEncode[T any](sch avro.Schema, version int, vals []T, out io.Writer) error {
enc, err := ocf.NewEncoderWithSchema(
- internal.AvroSchemaCache.Get(key),
- out, ocf.WithMetadata(map[string][]byte{
+ sch,
+ out,
+ ocf.WithMetadata(map[string][]byte{
"format-version": []byte(strconv.Itoa(version)),
}),
ocf.WithCodec(ocf.Deflate),
@@ -238,3 +239,68 @@ func avroEncode[T any](key string, version int, vals []T, out io.Writer) error {
return enc.Close()
}
+
+func structTypeToAvroPartitionSchema(st *StructType) (avro.Schema, error) {
+ var aFields []*avro.Field
+
+ for _, field := range st.Fields() {
+ aField, err := nestedFieldToAvroField(field)
+
+ if err != nil {
+ return nil, err
+ }
+
+ aFields = append(aFields, aField)
+ }
+
+ return avro.NewRecordSchema("r102", "", aFields)
+}
+
+func nestedFieldToAvroField(f NestedField) (*avro.Field, error) {
+ sch, err := nestedFieldToAvroSchema(f)
+
+ if err != nil {
+ return nil, err
+ }
+
+ return avro.NewField(
+ f.Name,
+ sch,
+ avro.WithDoc(f.Doc),
+ avro.WithProps(map[string]any{"field-id": f.ID}),
+ )
+}
+
+func nestedFieldToAvroSchema(f NestedField) (avro.Schema, error) {
+ var sch avro.Schema
+
+ switch f.Type.(type) {
+ case *StringType:
+ sch = avro.NewPrimitiveSchema(avro.String, nil)
+ case *Int32Type:
+ sch = avro.NewPrimitiveSchema(avro.Int, nil)
+ case *Int64Type:
+ sch = avro.NewPrimitiveSchema(avro.Long, nil)
+ case *BinaryType:
+ sch = avro.NewPrimitiveSchema(avro.Bytes, nil)
+ case *BooleanType:
+ sch = avro.NewPrimitiveSchema(avro.Boolean, nil)
+ case *Float32Type:
+ sch = avro.NewPrimitiveSchema(avro.Float, nil)
+ case *Float64Type:
+ sch = avro.NewPrimitiveSchema(avro.Double, nil)
+ case *DateType:
+ sch = avro.NewPrimitiveSchema(avro.Int, avro.NewPrimitiveLogicalSchema(avro.Date))
+ default:
+ return nil, fmt.Errorf("unsupported Iceberg type: %s", f.Type.String())
+ }
+
+ if !f.Required {
+ return avro.NewUnionSchema([]avro.Schema{
+ avro.NewNullSchema(),
+ sch,
+ })
+ }
+
+ return sch, nil
+}