Skip to content

Commit 93234f2

Browse files
manuzhangcodex
manuzhang
and
codex
committed
fix(inspect): address metadata table review findings
Co-authored-by: Codex <codex@openai.com>
1 parent 1744e8f commit 93234f2

10 files changed

Lines changed: 313 additions & 63 deletions

‎src/iceberg/cpp20_compatibility_internal.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,8 +75,12 @@
7575
#include "iceberg/iceberg_data_export.h"
7676
#include "iceberg/iceberg_export.h"
7777
#include "iceberg/inheritable_metadata.h"
78+
#include "iceberg/inspect/files_table.h"
7879
#include "iceberg/inspect/history_table.h"
80+
#include "iceberg/inspect/manifests_table.h"
7981
#include "iceberg/inspect/metadata_table.h"
82+
#include "iceberg/inspect/partitions_table.h"
83+
#include "iceberg/inspect/refs_table.h"
8084
#include "iceberg/inspect/snapshots_table.h"
8185
#include "iceberg/location_provider.h"
8286
#include "iceberg/logging/cerr_logger.h"

‎src/iceberg/inspect/files_table.cc‎

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -31,33 +31,37 @@
3131

3232
namespace iceberg {
3333

34-
FilesTable::FilesTable(std::shared_ptr<Table> table, std::shared_ptr<Schema> schema,
35-
std::shared_ptr<Schema> table_schema,
36-
std::shared_ptr<StructType> partition_type)
37-
: TimeTravelMetadataTable(std::move(table)),
38-
schema_(std::move(schema)),
39-
table_schema_(std::move(table_schema)),
40-
partition_type_(std::move(partition_type)) {}
34+
FilesTable::FilesTable(std::shared_ptr<Table> table)
35+
: TimeTravelMetadataTable(std::move(table)) {}
4136

4237
FilesTable::~FilesTable() = default;
4338

4439
const std::shared_ptr<Schema>& FilesTable::schema() const { return schema_; }
4540

4641
Result<std::unique_ptr<FilesTable>> FilesTable::Make(std::shared_ptr<Table> table) {
4742
ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
48-
ICEBERG_ASSIGN_OR_RAISE(auto table_schema, table->schema());
49-
ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(*table));
43+
auto files = std::unique_ptr<FilesTable>(new FilesTable(std::move(table)));
44+
ICEBERG_RETURN_UNEXPECTED(files->RefreshSchema());
45+
return files;
46+
}
47+
48+
Status FilesTable::RefreshSchema() {
49+
ICEBERG_ASSIGN_OR_RAISE(auto table_schema, source_table()->schema());
50+
ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
51+
internal::UnifiedPartitionType(*source_table()));
5052
ICEBERG_ASSIGN_OR_RAISE(auto schema,
5153
internal::FilesTableSchema(*table_schema, partition_type));
52-
return std::unique_ptr<FilesTable>(new FilesTable(std::move(table), std::move(schema),
53-
std::move(table_schema),
54-
std::move(partition_type)));
54+
schema_ = std::move(schema);
55+
table_schema_ = std::move(table_schema);
56+
partition_type_ = std::move(partition_type);
57+
return {};
5558
}
5659

5760
Result<ArrowArrayStream> FilesTable::ScanSnapshot(
5861
const SnapshotSelection& snapshot_selection) {
5962
ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
6063
*source_table(), snapshot_selection));
64+
ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
6165
ICEBERG_ASSIGN_OR_RAISE(auto files, internal::LiveFiles(*source_table(), snapshot));
6266
auto schema = schema_;
6367
auto table_schema = table_schema_;

‎src/iceberg/inspect/files_table.h‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -40,16 +40,19 @@ class ICEBERG_EXPORT FilesTable : public TimeTravelMetadataTable {
4040

4141
Kind kind() const noexcept override { return Kind::kFiles; }
4242

43+
/// \brief Schema from construction or the most recent scan.
44+
/// A new scan rebuilds this schema from the source table's current metadata.
45+
/// Previously returned streams retain their own schemas.
4346
const std::shared_ptr<Schema>& schema() const override;
4447

4548
protected:
4649
Result<ArrowArrayStream> ScanSnapshot(
4750
const SnapshotSelection& snapshot_selection) override;
4851

4952
private:
50-
FilesTable(std::shared_ptr<Table> table, std::shared_ptr<Schema> schema,
51-
std::shared_ptr<Schema> table_schema,
52-
std::shared_ptr<StructType> partition_type);
53+
explicit FilesTable(std::shared_ptr<Table> table);
54+
55+
Status RefreshSchema();
5356

5457
std::shared_ptr<Schema> schema_;
5558
std::shared_ptr<Schema> table_schema_;

‎src/iceberg/inspect/manifests_table.cc‎

Lines changed: 26 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,14 @@ Status AppendPartitionSummaries(ArrowArray* array, const ManifestRow& row,
106106
return {};
107107
}
108108

109+
Status AppendManifestCount(ArrowArray* array, const std::optional<int32_t>& count,
110+
bool applicable) {
111+
if (!applicable) {
112+
return AppendInt(array, 0);
113+
}
114+
return count.has_value() ? AppendInt(array, *count) : AppendNull(array);
115+
}
116+
109117
Status AppendManifest(ArrowRowBuilder& builder, const ManifestRow& row,
110118
const std::shared_ptr<Schema>& table_schema) {
111119
const auto& manifest = row.manifest;
@@ -114,21 +122,23 @@ Status AppendManifest(ArrowRowBuilder& builder, const ManifestRow& row,
114122
ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(1), manifest.manifest_path));
115123
ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), manifest.manifest_length));
116124
ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(3), manifest.partition_spec_id));
117-
ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(4), manifest.added_snapshot_id));
125+
ICEBERG_PRECHECK(manifest.added_snapshot_id.has_value(),
126+
"Manifest '{}' has no added snapshot ID", manifest.manifest_path);
127+
ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(4), *manifest.added_snapshot_id));
118128

119129
const bool data = manifest.content == ManifestContent::kData;
120130
ICEBERG_RETURN_UNEXPECTED(
121-
AppendInt(builder.column(5), data ? manifest.added_files_count.value_or(0) : 0));
131+
AppendManifestCount(builder.column(5), manifest.added_files_count, data));
122132
ICEBERG_RETURN_UNEXPECTED(
123-
AppendInt(builder.column(6), data ? manifest.existing_files_count.value_or(0) : 0));
133+
AppendManifestCount(builder.column(6), manifest.existing_files_count, data));
124134
ICEBERG_RETURN_UNEXPECTED(
125-
AppendInt(builder.column(7), data ? manifest.deleted_files_count.value_or(0) : 0));
135+
AppendManifestCount(builder.column(7), manifest.deleted_files_count, data));
126136
ICEBERG_RETURN_UNEXPECTED(
127-
AppendInt(builder.column(8), data ? 0 : manifest.added_files_count.value_or(0)));
137+
AppendManifestCount(builder.column(8), manifest.added_files_count, !data));
128138
ICEBERG_RETURN_UNEXPECTED(
129-
AppendInt(builder.column(9), data ? 0 : manifest.existing_files_count.value_or(0)));
139+
AppendManifestCount(builder.column(9), manifest.existing_files_count, !data));
130140
ICEBERG_RETURN_UNEXPECTED(
131-
AppendInt(builder.column(10), data ? 0 : manifest.deleted_files_count.value_or(0)));
141+
AppendManifestCount(builder.column(10), manifest.deleted_files_count, !data));
132142

133143
ICEBERG_ASSIGN_OR_RAISE(auto partition_type, row.spec->PartitionType(*table_schema));
134144
ICEBERG_RETURN_UNEXPECTED(AppendPartitionSummaries(
@@ -150,12 +160,13 @@ const std::shared_ptr<Schema>& ManifestsTable::schema() const {
150160
SchemaField::MakeRequired(2, "length", int64()),
151161
SchemaField::MakeRequired(3, "partition_spec_id", int32()),
152162
SchemaField::MakeRequired(4, "added_snapshot_id", int64()),
153-
SchemaField::MakeRequired(5, "added_data_files_count", int32()),
154-
SchemaField::MakeRequired(6, "existing_data_files_count", int32()),
155-
SchemaField::MakeRequired(7, "deleted_data_files_count", int32()),
156-
SchemaField::MakeRequired(15, "added_delete_files_count", int32()),
157-
SchemaField::MakeRequired(16, "existing_delete_files_count", int32()),
158-
SchemaField::MakeRequired(17, "deleted_delete_files_count", int32()),
163+
// Legacy manifests can omit counts; unknown counts must stay null.
164+
SchemaField::MakeOptional(5, "added_data_files_count", int32()),
165+
SchemaField::MakeOptional(6, "existing_data_files_count", int32()),
166+
SchemaField::MakeOptional(7, "deleted_data_files_count", int32()),
167+
SchemaField::MakeOptional(15, "added_delete_files_count", int32()),
168+
SchemaField::MakeOptional(16, "existing_delete_files_count", int32()),
169+
SchemaField::MakeOptional(17, "deleted_delete_files_count", int32()),
159170
SchemaField::MakeRequired(
160171
8, "partition_summaries",
161172
list(SchemaField::MakeRequired(
@@ -181,9 +192,9 @@ Result<ArrowArrayStream> ManifestsTable::ScanSnapshot(
181192
std::vector<ManifestRow> rows;
182193
if (snapshot != nullptr) {
183194
ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, source_table()->specs());
184-
SnapshotCache snapshot_cache(snapshot.get());
195+
SnapshotReader snapshot_reader(snapshot.get());
185196
ICEBERG_ASSIGN_OR_RAISE(auto manifests,
186-
snapshot_cache.Manifests(source_table()->io()));
197+
snapshot_reader.Manifests(source_table()->io()));
187198
rows.reserve(manifests.size());
188199
for (const auto& manifest : manifests) {
189200
auto spec = specs_ref.get().find(manifest.partition_spec_id);

‎src/iceberg/inspect/metadata_table_stream_internal.h‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -34,8 +34,8 @@
3434
#include "iceberg/nanoarrow_status_internal.h"
3535
#include "iceberg/schema.h"
3636
#include "iceberg/schema_internal.h"
37-
#include "iceberg/util/iterator.h"
3837
#include "iceberg/util/macros.h"
38+
#include "iceberg/util/stream.h"
3939

4040
namespace iceberg::internal {
4141

@@ -46,7 +46,7 @@ class MetadataTableRowsStream {
4646
using AppendRow = std::function<Status(ArrowRowBuilder&, const Row&)>;
4747

4848
static Result<std::unique_ptr<MetadataTableRowsStream>> Make(
49-
const Schema& schema, std::unique_ptr<Iterator<Row>> rows, AppendRow append_row) {
49+
const Schema& schema, std::unique_ptr<Stream<Row>> rows, AppendRow append_row) {
5050
ArrowSchema arrow_schema{};
5151
ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
5252
return std::unique_ptr<MetadataTableRowsStream>(new MetadataTableRowsStream(
@@ -96,20 +96,20 @@ class MetadataTableRowsStream {
9696
}
9797

9898
private:
99-
MetadataTableRowsStream(std::unique_ptr<Iterator<Row>> rows, AppendRow append_row,
99+
MetadataTableRowsStream(std::unique_ptr<Stream<Row>> rows, AppendRow append_row,
100100
ArrowSchema arrow_schema)
101101
: rows_(std::move(rows)),
102102
append_row_(std::move(append_row)),
103103
arrow_schema_(std::move(arrow_schema)) {}
104104

105-
std::unique_ptr<Iterator<Row>> rows_;
105+
std::unique_ptr<Stream<Row>> rows_;
106106
AppendRow append_row_;
107107
ArrowSchema arrow_schema_{};
108108
};
109109

110110
template <typename Row, typename AppendRow>
111111
Result<ArrowArrayStream> MakeMetadataTableStream(const Schema& schema,
112-
std::unique_ptr<Iterator<Row>> rows,
112+
std::unique_ptr<Stream<Row>> rows,
113113
AppendRow append_row) {
114114
ICEBERG_ASSIGN_OR_RAISE(
115115
auto stream,
@@ -120,7 +120,7 @@ Result<ArrowArrayStream> MakeMetadataTableStream(const Schema& schema,
120120
}
121121

122122
template <typename Row>
123-
class MetadataTableVectorIterator : public Iterator<Row> {
123+
class MetadataTableVectorIterator : public Stream<Row> {
124124
public:
125125
explicit MetadataTableVectorIterator(std::vector<Row> rows) : rows_(std::move(rows)) {}
126126

@@ -141,7 +141,7 @@ template <typename Row, typename AppendRow>
141141
Result<ArrowArrayStream> MakeMetadataTableStream(const Schema& schema,
142142
std::vector<Row> rows,
143143
AppendRow append_row) {
144-
std::unique_ptr<Iterator<Row>> iterator =
144+
std::unique_ptr<Stream<Row>> iterator =
145145
std::make_unique<MetadataTableVectorIterator<Row>>(std::move(rows));
146146
return MakeMetadataTableStream(schema, std::move(iterator), std::move(append_row));
147147
}

‎src/iceberg/inspect/metadata_table_util_internal.cc‎

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -71,8 +71,8 @@ Result<std::shared_ptr<Snapshot>> SnapshotAtRef(const Table& table,
7171
}
7272

7373
auto ref = metadata->refs.find(std::string(ref_name));
74-
ICEBERG_CHECK(ref != metadata->refs.end(), "Cannot find snapshot reference '{}'",
75-
ref_name);
74+
ICEBERG_PRECHECK(ref != metadata->refs.end(), "Cannot find snapshot reference '{}'",
75+
ref_name);
7676
ICEBERG_PRECHECK(ref->second != nullptr, "Snapshot reference '{}' is null", ref_name);
7777
return metadata->SnapshotById(ref->second->snapshot_id);
7878
}
@@ -548,15 +548,16 @@ Status AppendDataFile(ArrowRowBuilder& builder, const Schema& schema,
548548

549549
namespace {
550550

551-
class LiveFilesIterator : public Iterator<LiveFile> {
551+
class LiveFilesIterator : public Stream<LiveFile> {
552552
public:
553553
LiveFilesIterator(std::shared_ptr<FileIO> io, std::shared_ptr<Schema> schema,
554554
std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs,
555-
std::vector<ManifestFile> manifests)
555+
std::vector<ManifestFile> manifests, std::vector<std::string> columns)
556556
: io_(std::move(io)),
557557
schema_(std::move(schema)),
558558
specs_(std::move(specs)),
559-
manifests_(std::move(manifests)) {}
559+
manifests_(std::move(manifests)),
560+
columns_(std::move(columns)) {}
560561

561562
protected:
562563
Result<std::optional<LiveFile>> NextImpl() override {
@@ -576,6 +577,9 @@ class LiveFilesIterator : public Iterator<LiveFile> {
576577
current_spec_ = spec->second;
577578
ICEBERG_ASSIGN_OR_RAISE(auto reader,
578579
ManifestReader::Make(manifest, io_, schema_, specs_));
580+
if (!columns_.empty()) {
581+
reader->Select(columns_);
582+
}
579583
ICEBERG_ASSIGN_OR_RAISE(entries_, reader->LiveEntries());
580584
}
581585
auto& entry = entries_[entry_index_++];
@@ -591,6 +595,7 @@ class LiveFilesIterator : public Iterator<LiveFile> {
591595
std::shared_ptr<Schema> schema_;
592596
std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs_;
593597
std::vector<ManifestFile> manifests_;
598+
std::vector<std::string> columns_;
594599
std::vector<ManifestEntry> entries_;
595600
std::shared_ptr<PartitionSpec> current_spec_;
596601
size_t manifest_index_ = 0;
@@ -599,19 +604,21 @@ class LiveFilesIterator : public Iterator<LiveFile> {
599604

600605
} // namespace
601606

602-
Result<std::unique_ptr<Iterator<LiveFile>>> LiveFiles(
603-
const Table& table, const std::shared_ptr<Snapshot>& snapshot) {
607+
Result<std::unique_ptr<Stream<LiveFile>>> LiveFiles(
608+
const Table& table, const std::shared_ptr<Snapshot>& snapshot,
609+
std::vector<std::string> columns) {
604610
ICEBERG_ASSIGN_OR_RAISE(auto schema, table.schema());
605611
ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs());
606612
std::vector<ManifestFile> manifests;
607613
if (snapshot != nullptr) {
608-
SnapshotCache snapshot_cache(snapshot.get());
614+
SnapshotReader snapshot_reader(snapshot.get());
609615
ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests,
610-
snapshot_cache.Manifests(table.io()));
616+
snapshot_reader.Manifests(table.io()));
611617
manifests.assign(snapshot_manifests.begin(), snapshot_manifests.end());
612618
}
613619
return std::make_unique<LiveFilesIterator>(table.io(), std::move(schema),
614-
specs_ref.get(), std::move(manifests));
620+
specs_ref.get(), std::move(manifests),
621+
std::move(columns));
615622
}
616623

617624
} // namespace iceberg::internal

‎src/iceberg/inspect/metadata_table_util_internal.h‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,14 +22,15 @@
2222
#include <cstdint>
2323
#include <memory>
2424
#include <optional>
25+
#include <string>
2526
#include <vector>
2627

2728
#include "iceberg/inspect/metadata_table.h"
2829
#include "iceberg/manifest/manifest_entry.h"
2930
#include "iceberg/result.h"
3031
#include "iceberg/row/partition_values.h"
3132
#include "iceberg/type_fwd.h"
32-
#include "iceberg/util/iterator.h"
33+
#include "iceberg/util/stream.h"
3334

3435
namespace iceberg {
3536
class ArrowRowBuilder;
@@ -69,7 +70,10 @@ Status AppendDataFile(ArrowRowBuilder& builder, const Schema& schema,
6970
const LiveFile& live_file);
7071

7172
/// \brief Iterate over live files, reading one manifest at a time.
72-
Result<std::unique_ptr<Iterator<LiveFile>>> LiveFiles(
73-
const Table& table, const std::shared_ptr<Snapshot>& snapshot);
73+
/// An empty column selection reads all data-file fields. Each manifest's entries
74+
/// are materialized by ManifestReader, independently of the Arrow batch size.
75+
Result<std::unique_ptr<Stream<LiveFile>>> LiveFiles(
76+
const Table& table, const std::shared_ptr<Snapshot>& snapshot,
77+
std::vector<std::string> columns = {});
7478

7579
} // namespace iceberg::internal

‎src/iceberg/inspect/partitions_table.cc‎

Lines changed: 19 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -207,12 +207,8 @@ void UpdateCounts(PartitionStats& partition, const DataFile& file) {
207207

208208
} // namespace
209209

210-
PartitionsTable::PartitionsTable(std::shared_ptr<Table> table,
211-
std::shared_ptr<Schema> schema,
212-
std::shared_ptr<StructType> partition_type)
213-
: TimeTravelMetadataTable(std::move(table)),
214-
schema_(std::move(schema)),
215-
partition_type_(std::move(partition_type)) {}
210+
PartitionsTable::PartitionsTable(std::shared_ptr<Table> table)
211+
: TimeTravelMetadataTable(std::move(table)) {}
216212

217213
PartitionsTable::~PartitionsTable() = default;
218214

@@ -221,7 +217,15 @@ const std::shared_ptr<Schema>& PartitionsTable::schema() const { return schema_;
221217
Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
222218
std::shared_ptr<Table> table) {
223219
ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
224-
ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(*table));
220+
auto partitions =
221+
std::unique_ptr<PartitionsTable>(new PartitionsTable(std::move(table)));
222+
ICEBERG_RETURN_UNEXPECTED(partitions->RefreshSchema());
223+
return partitions;
224+
}
225+
226+
Status PartitionsTable::RefreshSchema() {
227+
ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
228+
internal::UnifiedPartitionType(*source_table()));
225229

226230
std::vector<SchemaField> fields;
227231
if (!partition_type->fields().empty()) {
@@ -247,16 +251,20 @@ Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
247251
fields.push_back(SchemaField::MakeOptional(kLastUpdatedSnapshotIdFieldId,
248252
"last_updated_snapshot_id", int64()));
249253

250-
auto schema = std::make_shared<Schema>(std::move(fields));
251-
return std::unique_ptr<PartitionsTable>(new PartitionsTable(
252-
std::move(table), std::move(schema), std::move(partition_type)));
254+
schema_ = std::make_shared<Schema>(std::move(fields));
255+
partition_type_ = std::move(partition_type);
256+
return {};
253257
}
254258

255259
Result<ArrowArrayStream> PartitionsTable::ScanSnapshot(
256260
const SnapshotSelection& snapshot_selection) {
257261
ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
258262
*source_table(), snapshot_selection));
259-
ICEBERG_ASSIGN_OR_RAISE(auto files, internal::LiveFiles(*source_table(), snapshot));
263+
ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
264+
ICEBERG_ASSIGN_OR_RAISE(
265+
auto files, internal::LiveFiles(
266+
*source_table(), snapshot,
267+
{"content", "partition", "record_count", "file_size_in_bytes"}));
260268

261269
std::vector<PartitionStats> partitions;
262270
std::unordered_map<PartitionKey, size_t, PartitionKeyHash> positions;

0 commit comments

Comments
 (0)