| /* |
| * 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/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, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| 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, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| 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, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| 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, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| 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{}}); |
| |
| { |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| 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, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| 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(), 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)); |
| } |
| } |
| |
| TEST_P(TableScanTest, IncludeColumnStatsRejectsMissingColumn) { |
| ICEBERG_UNWRAP_OR_FAIL(auto builder, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| 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, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| builder->MinRowsRequested(-1); |
| EXPECT_THAT(builder->Build(), IsError(ErrorKind::kValidationFailed)); |
| |
| // Test invalid snapshot ID |
| ICEBERG_UNWRAP_OR_FAIL(auto builder2, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| builder2->UseSnapshot(9999L); |
| EXPECT_THAT(builder2->Build(), IsError(ErrorKind::kValidationFailed)); |
| |
| // Test invalid ref |
| ICEBERG_UNWRAP_OR_FAIL(auto builder3, |
| DataTableScanBuilder::Make(table_metadata_, file_io_)); |
| builder3->UseRef("non-existent-ref"); |
| EXPECT_THAT(builder3->Build(), IsError(ErrorKind::kValidationFailed)); |
| |
| // Test null inputs |
| EXPECT_THAT(DataTableScanBuilder::Make(nullptr, file_io_), |
| IsError(ErrorKind::kInvalidArgument)); |
| EXPECT_THAT(DataTableScanBuilder::Make(table_metadata_, nullptr), |
| IsError(ErrorKind::kInvalidArgument)); |
| } |
| |
| 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, |
| DataTableScanBuilder::Make(empty_metadata, file_io_)); |
| ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); |
| ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); |
| EXPECT_TRUE(tasks.empty()); |
| } |
| |
| 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, |
| DataTableScanBuilder::Make(metadata_with_manifest, file_io_)); |
| 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_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, |
| DataTableScanBuilder::Make(metadata_with_manifests, file_io_)); |
| |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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)), |
| 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_); |
| |
| // Create delete manifest with position delete files |
| 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, |
| MakeEqualityDeleteFile("/path/to/eq_delete.parquet", part_value, |
| partitioned_spec_->spec_id(), {1}))}; |
| 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, |
| DataTableScanBuilder::Make(metadata_with_manifests, file_io_)); |
| 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")); |
| // Verify that delete files are associated with the tasks |
| for (const auto& task : tasks) { |
| EXPECT_GT(task->delete_files().size(), 0); |
| } |
| } |
| |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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, DataTableScanBuilder::Make(metadata, file_io_)); |
| 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); |
| } |
| } |
| |
| INSTANTIATE_TEST_SUITE_P(TableScanVersions, TableScanTest, testing::Values(1, 2, 3)); |
| |
| } // namespace iceberg |
