| /* | |
| * 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. | |
| */ | |
| #include <limits> | |
| #include <memory> | |
| #include <optional> | |
| #include <string> | |
| #include <unordered_map> | |
| #include <vector> | |
| #include <gmock/gmock.h> | |
| #include <gtest/gtest.h> | |
| #include "iceberg/data_table_scan.h" | |
| #include "iceberg/expression/expressions.h" | |
| #include "iceberg/snapshot.h" | |
| #include "iceberg/table_metadata.h" | |
| #include "iceberg/test/executor.h" | |
| #include "iceberg/test/scan_test_base.h" | |
| namespace iceberg { | |
| class TableScanTest : public ScanTestBase { | |
| protected: | |
| void SetUp() override { | |
| ScanTestBase::SetUp(); | |
| MakeTableMetadata(); | |
| } | |
| void MakeTableMetadata() { | |
| constexpr int64_t kSnapshotId = 1000L; | |
| constexpr int64_t kSequenceNumber = 1L; | |
| const TimePointMs kTimestampMs = | |
| TimePointMsFromUnixMs(1609459200000L); // 2021-01-01 00:00:00 UTC | |
| auto snapshot = std::make_shared<Snapshot>( | |
| Snapshot{.snapshot_id = kSnapshotId, | |
| .parent_snapshot_id = std::nullopt, | |
| .sequence_number = kSequenceNumber, | |
| .timestamp_ms = kTimestampMs, | |
| .manifest_list = "/tmp/metadata/snap-1000-1-manifest-list.avro", | |
| .schema_id = schema_->schema_id()}); | |
| table_metadata_ = std::make_shared<TableMetadata>( | |
| TableMetadata{.format_version = 2, | |
| .table_uuid = "test-table-uuid", | |
| .location = "/tmp/table", | |
| .last_sequence_number = kSequenceNumber, | |
| .last_updated_ms = kTimestampMs, | |
| .last_column_id = 2, | |
| .schemas = {schema_}, | |
| .current_schema_id = schema_->schema_id(), | |
| .partition_specs = {partitioned_spec_, unpartitioned_spec_}, | |
| .default_spec_id = partitioned_spec_->spec_id(), | |
| .last_partition_id = 1000, | |
| .current_snapshot_id = kSnapshotId, | |
| .snapshots = {snapshot}, | |
| .snapshot_log = {SnapshotLogEntry{.timestamp_ms = kTimestampMs, | |
| .snapshot_id = kSnapshotId}}, | |
| .default_sort_order_id = 0, | |
| .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = kSnapshotId, | |
| .retention = SnapshotRef::Branch{}})}}}); | |
| } | |
| std::shared_ptr<DataFile> MakePositionDeleteFile( | |
| const std::string& path, const PartitionValues& partition, int32_t spec_id, | |
| std::optional<std::string> referenced_file = std::nullopt) { | |
| return std::make_shared<DataFile>(DataFile{ | |
| .content = DataFile::Content::kPositionDeletes, | |
| .file_path = path, | |
| .file_format = FileFormatType::kParquet, | |
| .partition = partition, | |
| .record_count = 1, | |
| .file_size_in_bytes = 10, | |
| .referenced_data_file = referenced_file, | |
| .partition_spec_id = spec_id, | |
| }); | |
| } | |
| std::shared_ptr<DataFile> MakeEqualityDeleteFile(const std::string& path, | |
| const PartitionValues& partition, | |
| int32_t spec_id, | |
| std::vector<int> equality_ids = {1}) { | |
| return std::make_shared<DataFile>(DataFile{ | |
| .content = DataFile::Content::kEqualityDeletes, | |
| .file_path = path, | |
| .file_format = FileFormatType::kParquet, | |
| .partition = partition, | |
| .record_count = 1, | |
| .file_size_in_bytes = 10, | |
| .equality_ids = std::move(equality_ids), | |
| .partition_spec_id = spec_id, | |
| }); | |
| } | |
| std::shared_ptr<DataFile> MakeDataFile(const std::string& path, | |
| const PartitionValues& partition, | |
| int32_t spec_id, int64_t record_count = 1, | |
| std::optional<int32_t> lower_id = std::nullopt, | |
| std::optional<int32_t> upper_id = std::nullopt) { | |
| auto file = std::make_shared<DataFile>(DataFile{ | |
| .file_path = path, | |
| .file_format = FileFormatType::kParquet, | |
| .partition = partition, | |
| .record_count = record_count, | |
| .file_size_in_bytes = 10, | |
| .sort_order_id = 0, | |
| .partition_spec_id = spec_id, | |
| }); | |
| // Set lower/upper bounds for field_id=1 ("id" column) if provided | |
| if (lower_id.has_value()) { | |
| file->lower_bounds[1] = Literal::Int(lower_id.value()).Serialize().value(); | |
| } | |
| if (upper_id.has_value()) { | |
| file->upper_bounds[1] = Literal::Int(upper_id.value()).Serialize().value(); | |
| } | |
| return file; | |
| } | |
| std::string WriteManifestList(int8_t format_version, int64_t snapshot_id, | |
| int64_t sequence_number, | |
| const std::vector<ManifestFile>& manifests) { | |
| return ScanTestBase::WriteManifestList(format_version, snapshot_id, | |
| 0L /*parent_snapshot_id*/, sequence_number, | |
| manifests); | |
| } | |
| std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> GetSpecsById() { | |
| return {{partitioned_spec_->spec_id(), partitioned_spec_}, | |
| {unpartitioned_spec_->spec_id(), unpartitioned_spec_}}; | |
| } | |
| std::shared_ptr<TableMetadata> table_metadata_; | |
| }; | |
| TEST_P(TableScanTest, TableScanBuilderOptions) { | |
| // Test basic scan creation and default values | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| ICEBERG_UNWRAP_OR_FAIL(auto basic_scan, builder->Build()); | |
| EXPECT_NE(basic_scan, nullptr); | |
| EXPECT_EQ(basic_scan->metadata(), table_metadata_); | |
| EXPECT_EQ(basic_scan->io(), file_io_); | |
| EXPECT_TRUE(basic_scan->is_case_sensitive()); | |
| // Test all builder options with method chaining | |
| auto projected_schema = std::make_shared<Schema>( | |
| std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32())}); | |
| auto filter = Expressions::Equal("id", Literal::Int(42)); | |
| constexpr int64_t kMinRows = 1000; | |
| constexpr int64_t kSnapshotId = 1000L; | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder2, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder2->Option("key1", "value1") | |
| .Option("key2", "value2") | |
| .CaseSensitive(false) | |
| .Project(projected_schema) | |
| .Filter(filter) | |
| .IncludeColumnStats({"id", "data"}) | |
| .IgnoreResiduals() | |
| .MinRowsRequested(kMinRows) | |
| .UseSnapshot(kSnapshotId) | |
| .Build()); | |
| // Verify all options were set correctly | |
| ICEBERG_UNWRAP_OR_FAIL(auto schema, scan->schema()); | |
| EXPECT_EQ(schema, projected_schema); | |
| EXPECT_EQ(scan->filter(), filter); | |
| EXPECT_FALSE(scan->is_case_sensitive()); | |
| const auto& context = scan->context(); | |
| EXPECT_EQ(context.options.at("key1"), "value1"); | |
| EXPECT_EQ(context.options.at("key2"), "value2"); | |
| EXPECT_TRUE(context.return_column_stats); | |
| EXPECT_EQ(context.columns_to_keep_stats.size(), 2); | |
| EXPECT_TRUE(context.columns_to_keep_stats.contains(1)); // id field | |
| EXPECT_TRUE(context.columns_to_keep_stats.contains(2)); // data field | |
| EXPECT_TRUE(context.ignore_residuals); | |
| EXPECT_TRUE(context.min_rows_requested.has_value()); | |
| EXPECT_EQ(context.min_rows_requested.value(), kMinRows); | |
| EXPECT_TRUE(context.snapshot_id.has_value()); | |
| EXPECT_EQ(context.snapshot_id.value(), kSnapshotId); | |
| // Test UseRef separately | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder3, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder3->UseRef("main"); | |
| ICEBERG_UNWRAP_OR_FAIL(auto ref_scan, builder3->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto snapshot, ref_scan->snapshot()); | |
| EXPECT_EQ(snapshot->snapshot_id, 1000L); | |
| } | |
| TEST_P(TableScanTest, UseRefPreservesInt64SnapshotIds) { | |
| constexpr int64_t kLargeSnapshotId = | |
| static_cast<int64_t>(std::numeric_limits<int32_t>::max()) + 42; | |
| table_metadata_->snapshots.push_back(std::make_shared<Snapshot>( | |
| Snapshot{.snapshot_id = kLargeSnapshotId, | |
| .parent_snapshot_id = table_metadata_->current_snapshot_id, | |
| .sequence_number = 2L, | |
| .timestamp_ms = TimePointMsFromUnixMs(1609459201000L), | |
| .manifest_list = "/tmp/metadata/snap-large-2-manifest-list.avro", | |
| .schema_id = schema_->schema_id()})); | |
| table_metadata_->refs["branch-with-large-snapshot-id"] = std::make_shared<SnapshotRef>( | |
| SnapshotRef{.snapshot_id = kLargeSnapshotId, .retention = SnapshotRef::Branch{}}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder->UseRef("branch-with-large-snapshot-id"); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ASSERT_TRUE(scan->context().snapshot_id.has_value()); | |
| EXPECT_EQ(scan->context().snapshot_id.value(), kLargeSnapshotId); | |
| ICEBERG_UNWRAP_OR_FAIL(auto snapshot, scan->snapshot()); | |
| EXPECT_EQ(snapshot->snapshot_id, kLargeSnapshotId); | |
| } | |
| TEST_P(TableScanTest, IncludeColumnStatsUsesFinalSnapshotSchema) { | |
| constexpr int64_t kBaseSnapshotId = 1000L; | |
| constexpr int64_t kEvolvedSnapshotId = 2000L; | |
| constexpr int32_t kBaseIdFieldId = 1; | |
| constexpr int32_t kEvolvedIdFieldId = 10; | |
| constexpr int32_t kEvolvedDataFieldId = 11; | |
| constexpr int32_t kEvolvedSchemaId = 1; | |
| auto evolved_schema = std::make_shared<Schema>( | |
| std::vector<SchemaField>{ | |
| SchemaField::MakeRequired(kEvolvedIdFieldId, "id", int32()), | |
| SchemaField::MakeRequired(kEvolvedDataFieldId, "data", string())}, | |
| kEvolvedSchemaId); | |
| table_metadata_->schemas.push_back(evolved_schema); | |
| table_metadata_->last_column_id = kEvolvedDataFieldId; | |
| table_metadata_->snapshots.push_back(std::make_shared<Snapshot>( | |
| Snapshot{.snapshot_id = kEvolvedSnapshotId, | |
| .parent_snapshot_id = kBaseSnapshotId, | |
| .sequence_number = 2L, | |
| .timestamp_ms = TimePointMsFromUnixMs(1609459201000L), | |
| .manifest_list = "/tmp/metadata/snap-2000-2-manifest-list.avro", | |
| .schema_id = evolved_schema->schema_id()})); | |
| table_metadata_->refs["evolved-branch"] = std::make_shared<SnapshotRef>( | |
| SnapshotRef{.snapshot_id = kEvolvedSnapshotId, .retention = SnapshotRef::Branch{}}); | |
| table_metadata_->refs["evolved-tag"] = std::make_shared<SnapshotRef>( | |
| SnapshotRef{.snapshot_id = kEvolvedSnapshotId, .retention = SnapshotRef::Tag{}}); | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder->IncludeColumnStats({"id"}).UseSnapshot(kEvolvedSnapshotId); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan_schema, scan->schema()); | |
| EXPECT_EQ(scan_schema->schema_id(), evolved_schema->schema_id()); | |
| const auto& stats_fields = scan->context().columns_to_keep_stats; | |
| EXPECT_EQ(stats_fields.size(), 1); | |
| EXPECT_TRUE(stats_fields.contains(kEvolvedIdFieldId)); | |
| EXPECT_FALSE(stats_fields.contains(kBaseIdFieldId)); | |
| } | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder->IncludeColumnStats({"id"}).UseRef("evolved-tag"); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan_schema, scan->schema()); | |
| EXPECT_EQ(scan_schema->schema_id(), evolved_schema->schema_id()); | |
| const auto& stats_fields = scan->context().columns_to_keep_stats; | |
| EXPECT_EQ(stats_fields.size(), 1); | |
| EXPECT_TRUE(stats_fields.contains(kEvolvedIdFieldId)); | |
| EXPECT_FALSE(stats_fields.contains(kBaseIdFieldId)); | |
| } | |
| // Branches use the table schema. | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder->IncludeColumnStats({"id"}).UseRef("evolved-branch"); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan_schema, scan->schema()); | |
| EXPECT_EQ(scan_schema->schema_id(), schema_->schema_id()); | |
| const auto& stats_fields = scan->context().columns_to_keep_stats; | |
| EXPECT_EQ(stats_fields.size(), 1); | |
| EXPECT_TRUE(stats_fields.contains(kBaseIdFieldId)); | |
| EXPECT_FALSE(stats_fields.contains(kEvolvedIdFieldId)); | |
| } | |
| } | |
| TEST_P(TableScanTest, IncludeColumnStatsRejectsMissingColumn) { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder->IncludeColumnStats({"missing"}); | |
| EXPECT_THAT(builder->Build(), | |
| ::testing::AllOf(IsError(ErrorKind::kValidationFailed), | |
| HasErrorMessage("Cannot find stats column: missing"))); | |
| } | |
| TEST_P(TableScanTest, TableScanBuilderValidationErrors) { | |
| // Test negative min rows | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder->MinRowsRequested(-1); | |
| EXPECT_THAT(builder->Build(), IsError(ErrorKind::kValidationFailed)); | |
| // Test invalid snapshot ID | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder2, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder2->UseSnapshot(9999L); | |
| EXPECT_THAT(builder2->Build(), IsError(ErrorKind::kValidationFailed)); | |
| // Test invalid ref | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder3, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder3->UseRef("non-existent-ref"); | |
| EXPECT_THAT(builder3->Build(), IsError(ErrorKind::kValidationFailed)); | |
| // An explicit empty selection still conflicts with an explicit projection schema. | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder4, MakeScanBuilder<DataTableScan>(table_metadata_)); | |
| builder4->Select({}).Project(schema_); | |
| EXPECT_THAT(builder4->Build(), | |
| ::testing::AllOf( | |
| IsError(ErrorKind::kValidationFailed), | |
| HasErrorMessage("Cannot set projection schema and selected columns"))); | |
| } | |
| TEST_P(TableScanTest, DataTableScanPlanFilesEmpty) { | |
| auto empty_metadata = std::make_shared<TableMetadata>( | |
| TableMetadata{.format_version = 2, | |
| .schemas = {schema_}, | |
| .current_schema_id = schema_->schema_id(), | |
| .partition_specs = {unpartitioned_spec_}, | |
| .default_spec_id = unpartitioned_spec_->spec_id(), | |
| .current_snapshot_id = -1, | |
| .snapshots = {}, | |
| .refs = {}}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(empty_metadata)); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| EXPECT_TRUE(tasks.empty()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto next, stream->Next()); | |
| EXPECT_FALSE(next.has_value()); | |
| } | |
| TEST_P(TableScanTest, PlanFilesWithDataManifests) { | |
| auto version = GetParam(); | |
| constexpr int64_t kSnapshotId = 1000L; | |
| const auto part_value = PartitionValues({Literal::Int(0)}); | |
| std::vector<ManifestEntry> data_entries{ | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data1.parquet", part_value, | |
| partitioned_spec_->spec_id(), /*record_count=*/100)), | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data2.parquet", part_value, | |
| partitioned_spec_->spec_id(), /*record_count=*/200))}; | |
| auto data_manifest = | |
| WriteDataManifest(version, kSnapshotId, std::move(data_entries), partitioned_spec_); | |
| std::string manifest_list_path = | |
| WriteManifestList(version, kSnapshotId, /*sequence_number=*/1, {data_manifest}); | |
| // Create a snapshot that references this manifest list | |
| auto timestamp_ms = TimePointMsFromUnixMs(1609459200000L); | |
| auto snapshot_with_manifest = | |
| std::make_shared<Snapshot>(Snapshot{.snapshot_id = kSnapshotId, | |
| .parent_snapshot_id = std::nullopt, | |
| .sequence_number = 1L, | |
| .timestamp_ms = timestamp_ms, | |
| .manifest_list = manifest_list_path, | |
| .summary = {}, | |
| .schema_id = schema_->schema_id()}); | |
| auto metadata_with_manifest = std::make_shared<TableMetadata>( | |
| TableMetadata{.format_version = 2, | |
| .table_uuid = "test-table-uuid", | |
| .location = "/tmp/table", | |
| .last_sequence_number = 1L, | |
| .last_updated_ms = timestamp_ms, | |
| .last_column_id = 2, | |
| .schemas = {schema_}, | |
| .current_schema_id = schema_->schema_id(), | |
| .partition_specs = {partitioned_spec_, unpartitioned_spec_}, | |
| .default_spec_id = partitioned_spec_->spec_id(), | |
| .last_partition_id = 1000, | |
| .current_snapshot_id = kSnapshotId, | |
| .snapshots = {snapshot_with_manifest}, | |
| .snapshot_log = {SnapshotLogEntry{.timestamp_ms = timestamp_ms, | |
| .snapshot_id = kSnapshotId}}, | |
| .default_sort_order_id = 0, | |
| .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = kSnapshotId, | |
| .retention = SnapshotRef::Branch{}, | |
| })}}}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, | |
| MakeScanBuilder<DataTableScan>(metadata_with_manifest)); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| ASSERT_EQ(tasks.size(), 2); | |
| EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet", | |
| "/path/to/data2.parquet")); | |
| ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream()); | |
| scan.reset(); | |
| ICEBERG_UNWRAP_OR_FAIL(auto streamed_tasks, stream->ToVector()); | |
| ASSERT_EQ(streamed_tasks.size(), 2); | |
| EXPECT_THAT( | |
| GetPaths(streamed_tasks), | |
| testing::UnorderedElementsAre("/path/to/data1.parquet", "/path/to/data2.parquet")); | |
| } | |
| TEST_P(TableScanTest, PlanRowLineage) { | |
| auto version = GetParam(); | |
| if (version < 3) { | |
| GTEST_SKIP() << "row lineage is only assigned in v3 manifests"; | |
| } | |
| constexpr int64_t kSnapshotId = 1000L; | |
| constexpr int64_t kDataSequenceNumber = 5L; | |
| constexpr int64_t kFileSequenceNumber = 7L; | |
| std::vector<ManifestEntry> data_entries{MakeEntry( | |
| ManifestStatus::kAdded, kSnapshotId, kDataSequenceNumber, | |
| MakeDataFile("/path/to/data.parquet", PartitionValues(std::vector<Literal>{}), | |
| unpartitioned_spec_->spec_id(), /*record_count=*/3))}; | |
| data_entries[0].file_sequence_number = kFileSequenceNumber; | |
| auto data_manifest = WriteDataManifest(version, kSnapshotId, std::move(data_entries), | |
| unpartitioned_spec_); | |
| std::string manifest_list_path = | |
| WriteManifestList(version, kSnapshotId, kFileSequenceNumber, {data_manifest}); | |
| auto timestamp_ms = TimePointMsFromUnixMs(1609459200000L); | |
| auto snapshot = | |
| std::make_shared<Snapshot>(Snapshot{.snapshot_id = kSnapshotId, | |
| .parent_snapshot_id = std::nullopt, | |
| .sequence_number = kFileSequenceNumber, | |
| .timestamp_ms = timestamp_ms, | |
| .manifest_list = manifest_list_path, | |
| .schema_id = schema_->schema_id(), | |
| .first_row_id = 0L, | |
| .added_rows = 3L}); | |
| auto metadata = ScanTestBase::MakeTableMetadata( | |
| {snapshot}, kSnapshotId, | |
| {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = kSnapshotId, .retention = SnapshotRef::Branch{}})}}, | |
| unpartitioned_spec_); | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| ASSERT_EQ(tasks.size(), 1); | |
| EXPECT_EQ(tasks[0]->data_file()->first_row_id, 0L); | |
| EXPECT_EQ(tasks[0]->data_file()->data_sequence_number, kDataSequenceNumber); | |
| } | |
| TEST_P(TableScanTest, PlanFilesWithMultipleManifests) { | |
| auto version = GetParam(); | |
| const auto partition_a = PartitionValues({Literal::Int(0)}); | |
| const auto partition_b = PartitionValues({Literal::Int(1)}); | |
| // Create first data manifest | |
| std::vector<ManifestEntry> data_entries_1{MakeEntry( | |
| ManifestStatus::kAdded, /*snapshot_id=*/1000L, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data1.parquet", partition_a, partitioned_spec_->spec_id()))}; | |
| auto data_manifest_1 = WriteDataManifest(version, /*snapshot_id=*/1000L, | |
| std::move(data_entries_1), partitioned_spec_); | |
| // Create second data manifest | |
| std::vector<ManifestEntry> data_entries_2{MakeEntry( | |
| ManifestStatus::kAdded, /*snapshot_id=*/1000L, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data2.parquet", partition_b, partitioned_spec_->spec_id()))}; | |
| auto data_manifest_2 = WriteDataManifest(version, /*snapshot_id=*/1000L, | |
| std::move(data_entries_2), partitioned_spec_); | |
| // Write manifest list with multiple manifests | |
| std::string manifest_list_path = | |
| WriteManifestList(version, /*snapshot_id=*/1000L, /*sequence_number=*/1, | |
| {data_manifest_1, data_manifest_2}); | |
| // Create a snapshot that references this manifest list | |
| auto timestamp_ms = TimePointMsFromUnixMs(1609459200000L); | |
| auto snapshot_with_manifests = | |
| std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1000L, | |
| .parent_snapshot_id = std::nullopt, | |
| .sequence_number = 1L, | |
| .timestamp_ms = timestamp_ms, | |
| .manifest_list = manifest_list_path, | |
| .summary = {}, | |
| .schema_id = schema_->schema_id()}); | |
| auto metadata_with_manifests = std::make_shared<TableMetadata>( | |
| TableMetadata{.format_version = 2, | |
| .table_uuid = "test-table-uuid", | |
| .location = "/tmp/table", | |
| .last_sequence_number = 1L, | |
| .last_updated_ms = timestamp_ms, | |
| .last_column_id = 2, | |
| .schemas = {schema_}, | |
| .current_schema_id = schema_->schema_id(), | |
| .partition_specs = {partitioned_spec_, unpartitioned_spec_}, | |
| .default_spec_id = partitioned_spec_->spec_id(), | |
| .last_partition_id = 1000, | |
| .current_snapshot_id = 1000L, | |
| .snapshots = {snapshot_with_manifests}, | |
| .snapshot_log = {SnapshotLogEntry{.timestamp_ms = timestamp_ms, | |
| .snapshot_id = 1000L}}, | |
| .default_sort_order_id = 0, | |
| .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = 1000L, | |
| .retention = SnapshotRef::Branch{}, | |
| })}}}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, | |
| MakeScanBuilder<DataTableScan>(metadata_with_manifests)); | |
| test::ThreadExecutor executor; | |
| builder->PlanWith(executor); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| ASSERT_EQ(tasks.size(), 2); | |
| EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet", | |
| "/path/to/data2.parquet")); | |
| EXPECT_EQ(executor.submit_count(), 2); | |
| } | |
| TEST_P(TableScanTest, PlanFilesWithFilter) { | |
| auto version = GetParam(); | |
| constexpr int64_t kSnapshotId = 1000L; | |
| const auto part_value = PartitionValues({Literal::Int(0)}); | |
| // Create two data files with non-overlapping id ranges: | |
| // - data1.parquet: id range [1, 50] | |
| // - data2.parquet: id range [51, 100] | |
| std::vector<ManifestEntry> data_entries{ | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data1.parquet", part_value, | |
| partitioned_spec_->spec_id(), /*record_count=*/1, | |
| /*lower_id=*/1, /*upper_id=*/50)), | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data2.parquet", part_value, | |
| partitioned_spec_->spec_id(), /*record_count=*/1, | |
| /*lower_id=*/51, /*upper_id=*/100))}; | |
| auto data_manifest = | |
| WriteDataManifest(version, kSnapshotId, std::move(data_entries), partitioned_spec_); | |
| std::string manifest_list_path = | |
| WriteManifestList(version, kSnapshotId, /*sequence_number=*/1, {data_manifest}); | |
| auto timestamp_ms = TimePointMsFromUnixMs(1609459200000L); | |
| auto snapshot = std::make_shared<Snapshot>(Snapshot{.snapshot_id = kSnapshotId, | |
| .parent_snapshot_id = std::nullopt, | |
| .sequence_number = 1L, | |
| .timestamp_ms = timestamp_ms, | |
| .manifest_list = manifest_list_path, | |
| .schema_id = schema_->schema_id()}); | |
| auto metadata = std::make_shared<TableMetadata>(TableMetadata{ | |
| .format_version = 2, | |
| .table_uuid = "test-table-uuid", | |
| .location = "/tmp/table", | |
| .last_sequence_number = 1L, | |
| .last_updated_ms = timestamp_ms, | |
| .last_column_id = 2, | |
| .schemas = {schema_}, | |
| .current_schema_id = schema_->schema_id(), | |
| .partition_specs = {partitioned_spec_, unpartitioned_spec_}, | |
| .default_spec_id = partitioned_spec_->spec_id(), | |
| .last_partition_id = 1000, | |
| .current_snapshot_id = kSnapshotId, | |
| .snapshots = {snapshot}, | |
| .snapshot_log = {SnapshotLogEntry{.timestamp_ms = timestamp_ms, | |
| .snapshot_id = kSnapshotId}}, | |
| .default_sort_order_id = 0, | |
| .refs = {{"main", | |
| std::make_shared<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = kSnapshotId, .retention = SnapshotRef::Branch{}})}}}); | |
| // Test 1: Filter matches only data1.parquet (id=25 is in range [1, 50]) | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Filter(Expressions::Equal("id", Literal::Int(25))); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| ASSERT_EQ(tasks.size(), 1); | |
| EXPECT_EQ(tasks[0]->data_file()->file_path, "/path/to/data1.parquet"); | |
| } | |
| // Test 2: Filter matches only data2.parquet (id=75 is in range [51, 100]) | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Filter(Expressions::Equal("id", Literal::Int(75))); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| ASSERT_EQ(tasks.size(), 1); | |
| EXPECT_EQ(tasks[0]->data_file()->file_path, "/path/to/data2.parquet"); | |
| } | |
| // Test 3: Filter matches both files (id > 0 covers both ranges) | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Filter(Expressions::GreaterThan("id", Literal::Int(0))); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| ASSERT_EQ(tasks.size(), 2); | |
| EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet", | |
| "/path/to/data2.parquet")); | |
| } | |
| // Test 4: Filter matches no files (id=200 is outside both ranges) | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Filter(Expressions::Equal("id", Literal::Int(200))); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| EXPECT_TRUE(tasks.empty()); | |
| } | |
| } | |
| TEST_P(TableScanTest, PlanFilesWithDeleteFiles) { | |
| auto version = GetParam(); | |
| if (version < 2) { | |
| GTEST_SKIP() << "Delete files only supported in V2+"; | |
| } | |
| constexpr int64_t kSnapshotId = 1000L; | |
| const auto part_value = PartitionValues({Literal::Int(0)}); | |
| // Create data manifest with files | |
| std::vector<ManifestEntry> data_entries{ | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data1.parquet", part_value, | |
| partitioned_spec_->spec_id(), /*record_count=*/100, | |
| /*lower_id=*/0, /*upper_id=*/10)), | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1, | |
| MakeDataFile("/path/to/data2.parquet", part_value, | |
| partitioned_spec_->spec_id(), /*record_count=*/200, | |
| /*lower_id=*/20, /*upper_id=*/30))}; | |
| auto data_manifest = | |
| WriteDataManifest(version, kSnapshotId, std::move(data_entries), partitioned_spec_); | |
| // Create delete manifest with position delete files | |
| auto equality_delete = MakeEqualityDeleteFile("/path/to/eq_delete.parquet", part_value, | |
| partitioned_spec_->spec_id(), {1}); | |
| equality_delete->lower_bounds[1] = Literal::Int(20).Serialize().value(); | |
| equality_delete->upper_bounds[1] = Literal::Int(30).Serialize().value(); | |
| const auto other_part_value = PartitionValues({Literal::Int(1)}); | |
| auto cross_part_equality_delete = | |
| MakeEqualityDeleteFile("/path/to/cross_part_eq_delete.parquet", other_part_value, | |
| partitioned_spec_->spec_id(), {1}); | |
| cross_part_equality_delete->lower_bounds[1] = Literal::Int(0).Serialize().value(); | |
| cross_part_equality_delete->upper_bounds[1] = Literal::Int(100).Serialize().value(); | |
| std::vector<ManifestEntry> delete_entries{ | |
| MakeEntry( | |
| ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/2, | |
| MakePositionDeleteFile("/path/to/pos_delete.parquet", part_value, | |
| partitioned_spec_->spec_id(), "/path/to/data1.parquet")), | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/2, | |
| std::move(equality_delete)), | |
| MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/2, | |
| std::move(cross_part_equality_delete))}; | |
| auto delete_manifest = WriteDeleteManifest( | |
| version, kSnapshotId, std::move(delete_entries), partitioned_spec_); | |
| std::string manifest_list_path = WriteManifestList( | |
| version, kSnapshotId, /*sequence_number=*/2, {data_manifest, delete_manifest}); | |
| // Create a snapshot that references this manifest list | |
| auto timestamp_ms = TimePointMsFromUnixMs(1609459200000L); | |
| auto snapshot_with_manifests = | |
| std::make_shared<Snapshot>(Snapshot{.snapshot_id = kSnapshotId, | |
| .parent_snapshot_id = std::nullopt, | |
| .sequence_number = 2L, | |
| .timestamp_ms = timestamp_ms, | |
| .manifest_list = manifest_list_path, | |
| .summary = {}, | |
| .schema_id = schema_->schema_id()}); | |
| auto metadata_with_manifests = std::make_shared<TableMetadata>( | |
| TableMetadata{.format_version = 2, | |
| .table_uuid = "test-table-uuid", | |
| .location = "/tmp/table", | |
| .last_sequence_number = 2L, | |
| .last_updated_ms = timestamp_ms, | |
| .last_column_id = 2, | |
| .schemas = {schema_}, | |
| .current_schema_id = schema_->schema_id(), | |
| .partition_specs = {partitioned_spec_, unpartitioned_spec_}, | |
| .default_spec_id = partitioned_spec_->spec_id(), | |
| .last_partition_id = 1000, | |
| .current_snapshot_id = kSnapshotId, | |
| .snapshots = {snapshot_with_manifests}, | |
| .snapshot_log = {SnapshotLogEntry{.timestamp_ms = timestamp_ms, | |
| .snapshot_id = kSnapshotId}}, | |
| .default_sort_order_id = 0, | |
| .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = kSnapshotId, | |
| .retention = SnapshotRef::Branch{}, | |
| })}}}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, | |
| MakeScanBuilder<DataTableScan>(metadata_with_manifests)); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); | |
| auto verify_tasks = [](const auto& planned_tasks) { | |
| ASSERT_EQ(planned_tasks.size(), 2); | |
| for (const auto& task : planned_tasks) { | |
| ASSERT_EQ(task->delete_files().size(), 1); | |
| if (task->data_file()->file_path == "/path/to/data1.parquet") { | |
| EXPECT_EQ(task->delete_files().front()->file_path, "/path/to/pos_delete.parquet"); | |
| } else { | |
| EXPECT_EQ(task->data_file()->file_path, "/path/to/data2.parquet"); | |
| EXPECT_EQ(task->delete_files().front()->file_path, "/path/to/eq_delete.parquet"); | |
| } | |
| EXPECT_TRUE(task->data_file()->lower_bounds.empty()); | |
| EXPECT_TRUE(task->data_file()->upper_bounds.empty()); | |
| } | |
| }; | |
| verify_tasks(tasks); | |
| ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto streamed_tasks, stream->ToVector()); | |
| verify_tasks(streamed_tasks); | |
| } | |
| TEST_P(TableScanTest, SchemaWithSelectedColumnsAndFilter) { | |
| auto schema = std::make_shared<Schema>(std::vector<SchemaField>{ | |
| SchemaField::MakeRequired(/*field_id=*/1, "id", int32()), | |
| SchemaField::MakeRequired(/*field_id=*/2, "data", string()), | |
| SchemaField::MakeRequired(/*field_id=*/3, "value", int64())}); | |
| auto timestamp_ms = TimePointMsFromUnixMs(1609459200000L); | |
| auto metadata = std::make_shared<TableMetadata>(TableMetadata{ | |
| .format_version = 2, | |
| .table_uuid = "test-table-uuid", | |
| .location = "/tmp/table", | |
| .last_sequence_number = 1L, | |
| .last_updated_ms = timestamp_ms, | |
| .last_column_id = 3, | |
| .schemas = {schema}, | |
| .current_schema_id = schema->schema_id(), | |
| .partition_specs = {unpartitioned_spec_}, | |
| .default_spec_id = unpartitioned_spec_->spec_id(), | |
| .last_partition_id = 1000, | |
| .current_snapshot_id = 1000L, | |
| .snapshots = {std::make_shared<Snapshot>(Snapshot{ | |
| .snapshot_id = 1000L, | |
| .parent_snapshot_id = std::nullopt, | |
| .sequence_number = 1L, | |
| .timestamp_ms = timestamp_ms, | |
| .manifest_list = "/tmp/metadata/snap-1000-1-manifest-list.avro", | |
| .schema_id = schema->schema_id(), | |
| })}, | |
| .snapshot_log = {SnapshotLogEntry{.timestamp_ms = timestamp_ms, | |
| .snapshot_id = 1000L}}, | |
| .default_sort_order_id = 0, | |
| .refs = {{"main", std::make_shared<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = 1000L, | |
| .retention = SnapshotRef::Branch{}, | |
| })}}, | |
| }); | |
| // Select "data" column, filter on "id" column | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Select({"data"}).Filter(Expressions::Equal("id", Literal::Int(42))); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto projected_schema, scan->schema()); | |
| ASSERT_EQ(projected_schema->fields().size(), 2); | |
| ICEBERG_UNWRAP_OR_FAIL(auto id_field, projected_schema->FindFieldByName("id")); | |
| EXPECT_TRUE(id_field.has_value()); | |
| EXPECT_EQ(id_field->get().field_id(), 1); | |
| ICEBERG_UNWRAP_OR_FAIL(auto data_field, projected_schema->FindFieldByName("data")); | |
| EXPECT_TRUE(data_field.has_value()); | |
| EXPECT_EQ(data_field->get().field_id(), 2); | |
| } | |
| // Select "id" and "value", filter on "data" | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Select({"id", "value"}) | |
| .Filter(Expressions::Equal("data", Literal::String("test"))); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto projected_schema, scan->schema()); | |
| ASSERT_EQ(projected_schema->fields().size(), 3); | |
| ICEBERG_UNWRAP_OR_FAIL(auto id_field, projected_schema->FindFieldByName("id")); | |
| EXPECT_TRUE(id_field.has_value()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto data_field, projected_schema->FindFieldByName("data")); | |
| EXPECT_TRUE(data_field.has_value()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto value_field, projected_schema->FindFieldByName("value")); | |
| EXPECT_TRUE(value_field.has_value()); | |
| } | |
| // Select "id", filter on "id" - should only have "id" once | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Select({"id"}).Filter(Expressions::Equal("id", Literal::Int(42))); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto projected_schema, scan->schema()); | |
| ASSERT_EQ(projected_schema->fields().size(), 1); | |
| ICEBERG_UNWRAP_OR_FAIL(auto id_field, projected_schema->FindFieldByName("id")); | |
| EXPECT_TRUE(id_field.has_value()); | |
| EXPECT_EQ(id_field->get().field_id(), 1); | |
| } | |
| // Select columns without filter | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Select({"data"}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto projected_schema, scan->schema()); | |
| ASSERT_EQ(projected_schema->fields().size(), 1); | |
| ICEBERG_UNWRAP_OR_FAIL(auto data_field, projected_schema->FindFieldByName("data")); | |
| EXPECT_TRUE(data_field.has_value()); | |
| EXPECT_EQ(data_field->get().field_id(), 2); | |
| } | |
| // An explicit empty selection is different from the default projection. | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Select({}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto projected_schema, scan->schema()); | |
| EXPECT_TRUE(projected_schema->fields().empty()); | |
| } | |
| // A wildcard selection explicitly projects the full table schema. | |
| { | |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata)); | |
| builder->Select({std::string(Schema::kAllColumns)}); | |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); | |
| ICEBERG_UNWRAP_OR_FAIL(auto projected_schema, scan->schema()); | |
| EXPECT_EQ(*projected_schema, *schema); | |
| } | |
| } | |
| INSTANTIATE_TEST_SUITE_P(TableScanVersions, TableScanTest, testing::Values(1, 2, 3)); | |
| } // namespace iceberg |