src/iceberg/table_scan.cc

/*
 * 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 "iceberg/table_scan.h"

#include <algorithm>
#include <cstdint>
#include <iterator>
#include <ranges>
#include <utility>

#include "iceberg/data_table_scan.h"
#include "iceberg/expression/binder.h"
#include "iceberg/expression/expression.h"
#include "iceberg/incremental_append_scan.h"
#include "iceberg/incremental_changelog_scan.h"
#include "iceberg/manifest/manifest_entry.h"
#include "iceberg/metrics/metrics_reporters.h"
#include "iceberg/result.h"
#include "iceberg/schema.h"
#include "iceberg/snapshot.h"
#include "iceberg/table.h"
#include "iceberg/table_metadata.h"
#include "iceberg/util/content_file_util.h"
#include "iceberg/util/int128.h"
#include "iceberg/util/macros.h"
#include "iceberg/util/snapshot_util.h"
#include "iceberg/util/timepoint.h"
#include "iceberg/util/type_util.h"

namespace iceberg {

namespace {

const std::vector<std::string> kScanColumns = {
    "snapshot_id",         "file_path",          "file_ordinal",  "file_format",
    "block_size_in_bytes", "file_size_in_bytes", "record_count",  "partition",
    "key_metadata",        "split_offsets",      "sort_order_id",
};

const std::vector<std::string> kStatsColumns = {
    "value_counts", "null_value_counts", "nan_value_counts",
    "lower_bounds", "upper_bounds",      "column_sizes",
};

const std::vector<std::string> kScanColumnsWithStats = [] {
  auto cols = kScanColumns;
  cols.insert(cols.end(), kStatsColumns.begin(), kStatsColumns.end());
  return cols;
}();

}  // namespace

namespace internal {

Status TableScanContext::Validate() const {
  ICEBERG_CHECK(columns_to_keep_stats.empty() || return_column_stats,
                "Cannot select columns to keep stats when column stats are not returned");
  ICEBERG_CHECK(projected_schema == nullptr || !selected_columns.has_value(),
                "Cannot set projection schema and selected columns at the same time");
  ICEBERG_CHECK(!snapshot_id.has_value() ||
                    (!from_snapshot_id.has_value() && !to_snapshot_id.has_value()),
                "Cannot mix snapshot scan and incremental scan");
  ICEBERG_CHECK(!min_rows_requested.has_value() || min_rows_requested.value() >= 0,
                "Min rows requested cannot be negative");
  return {};
}

bool IsScanCurrentLineage(const TableScanContext& context) {
  return !context.from_snapshot_id.has_value() && !context.to_snapshot_id.has_value();
}

Result<int64_t> ToSnapshotIdInclusive(const TableScanContext& context,
                                      const TableMetadata& metadata) {
  // Get the branch's current snapshot ID if branch is set
  std::shared_ptr<Snapshot> branch_snapshot;
  const std::string& branch = context.branch;
  if (!branch.empty()) {
    auto iter = metadata.refs.find(branch);
    ICEBERG_CHECK(iter != metadata.refs.end() && iter->second != nullptr,
                  "Cannot find branch: {}", branch);
    ICEBERG_ASSIGN_OR_RAISE(branch_snapshot,
                            metadata.SnapshotById(iter->second->snapshot_id));
  }

  if (context.to_snapshot_id.has_value()) {
    int64_t to_snapshot_id_value = context.to_snapshot_id.value();

    if (branch_snapshot != nullptr) {
      // Validate `to_snapshot_id` is on the current branch
      ICEBERG_ASSIGN_OR_RAISE(
          bool is_ancestor,
          SnapshotUtil::IsAncestorOf(metadata, branch_snapshot->snapshot_id,
                                     to_snapshot_id_value));
      ICEBERG_CHECK(is_ancestor,
                    "End snapshot is not a valid snapshot on the current branch: {}",
                    branch);
    }

    return to_snapshot_id_value;
  }

  // If to_snapshot_id is not set, use branch's current snapshot if branch is set
  if (branch_snapshot != nullptr) {
    return branch_snapshot->snapshot_id;
  }

  // Get current snapshot from table's current snapshot
  std::shared_ptr<Snapshot> current_snapshot;
  ICEBERG_ASSIGN_OR_RAISE(current_snapshot, metadata.Snapshot());
  ICEBERG_CHECK(current_snapshot != nullptr,
                "End snapshot is not set and table has no current snapshot");
  return current_snapshot->snapshot_id;
}

Result<std::optional<int64_t>> FromSnapshotIdExclusive(const TableScanContext& context,
                                                       const TableMetadata& metadata,
                                                       int64_t to_snapshot_id_inclusive) {
  if (!context.from_snapshot_id.has_value()) {
    return std::nullopt;
  }

  int64_t from_snapshot_id = context.from_snapshot_id.value();

  // Validate `from_snapshot_id` is an ancestor of `to_snapshot_id_inclusive`
  if (context.from_snapshot_id_inclusive) {
    ICEBERG_ASSIGN_OR_RAISE(
        bool is_ancestor,
        SnapshotUtil::IsAncestorOf(metadata, to_snapshot_id_inclusive, from_snapshot_id));
    ICEBERG_CHECK(
        is_ancestor,
        "Starting snapshot (inclusive) {} is not an ancestor of end snapshot {}",
        from_snapshot_id, to_snapshot_id_inclusive);

    // For inclusive behavior, return the parent snapshot ID (can be nullopt)
    ICEBERG_ASSIGN_OR_RAISE(auto from_snapshot, metadata.SnapshotById(from_snapshot_id));
    return from_snapshot->parent_snapshot_id;
  }

  // Validate there is an ancestor of `to_snapshot_id_inclusive` where parent is
  // `from_snapshot_id`
  ICEBERG_ASSIGN_OR_RAISE(bool is_parent_ancestor,
                          SnapshotUtil::IsParentAncestorOf(
                              metadata, to_snapshot_id_inclusive, from_snapshot_id));
  ICEBERG_CHECK(
      is_parent_ancestor,
      "Starting snapshot (exclusive) {} is not a parent ancestor of end snapshot {}",
      from_snapshot_id, to_snapshot_id_inclusive);

  return from_snapshot_id;
}

}  // namespace internal

ScanTask::~ScanTask() = default;

// FileScanTask implementation

FileScanTask::FileScanTask(std::shared_ptr<DataFile> data_file,
                           std::vector<std::shared_ptr<DataFile>> delete_files,
                           std::shared_ptr<Expression> residual_filter)
    : data_file_(std::move(data_file)),
      delete_files_(std::move(delete_files)),
      residual_filter_(std::move(residual_filter)) {
  ICEBERG_DCHECK(data_file_ != nullptr, "Data file cannot be null for FileScanTask");
}

FileScanTask::FileScanTask(std::shared_ptr<DataFile> data_file,
                           std::vector<std::shared_ptr<DataFile>> delete_files,
                           std::shared_ptr<Expression> residual_filter, Range range)
    : FileScanTask(std::move(data_file), std::move(delete_files),
                   std::move(residual_filter)) {
  range_ = range;
}

Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
    std::shared_ptr<DataFile> data_file, int64_t start, int64_t length,
    std::vector<std::shared_ptr<DataFile>> delete_files,
    std::shared_ptr<Expression> filter) {
  if (data_file == nullptr) {
    return InvalidArgument("Cannot create a file scan task without a data file");
  }

  const int64_t file_size = data_file->file_size_in_bytes;
  if (file_size < 0) {
    return InvalidArgument("Cannot split file {} with negative size {}",
                           data_file->file_path, file_size);
  }
  if (start < 0 || start > file_size) {
    return InvalidArgument("Split start {} is outside file {} of size {}", start,
                           data_file->file_path, file_size);
  }
  // Subtraction is safe after validating start, even when file_size is INT64_MAX.
  if (length < 0 || length > file_size - start) {
    return InvalidArgument("Split length {} from start {} is outside file {} of size {}",
                           length, start, data_file->file_path, file_size);
  }
  // A full-file range is a whole-file task (is_split() is false). This is also the
  // only valid empty range: [0, 0) of an empty file.
  if (start == 0 && length == file_size) {
    return std::make_shared<FileScanTask>(std::move(data_file), std::move(delete_files),
                                          std::move(filter));
  }
  if (length == 0) {
    return InvalidArgument("A partial split of file {} must have positive length",
                           data_file->file_path);
  }

  return std::shared_ptr<FileScanTask>(
      new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter),
                       Range{.start = start, .length = length}));
}

int64_t FileScanTask::length() const {
  return range_ ? range_->length : data_file_->file_size_in_bytes;
}

int64_t FileScanTask::size_bytes() const { return length(); }

int32_t FileScanTask::files_count() const { return 1; }

int64_t FileScanTask::estimated_row_count() const {
  if (!range_) {
    return data_file_->record_count;
  }

  const int64_t record_count = data_file_->record_count;
  const int64_t file_size = data_file_->file_size_in_bytes;
  if (record_count <= 0 || file_size <= 0 || range_->start > file_size ||
      range_->length > file_size - range_->start) {
    // Keep arithmetic safe even for tasks invalidated by metadata mutation.
    return 0;
  }

  // floor(end * rows / size) - floor(start * rows / size). For non-overlapping
  // splits covering a file with unchanged metadata, estimates sum to record_count.
  // int128 avoids overflow.
  const auto end = static_cast<int128_t>(range_->start) + range_->length;
  const auto count_at_end = end * record_count / file_size;
  const auto count_at_start =
      static_cast<int128_t>(range_->start) * record_count / file_size;
  return static_cast<int64_t>(count_at_end - count_at_start);
}

// ChangelogScanTask implementation

ChangelogScanTask::ChangelogScanTask(int32_t change_ordinal, int64_t commit_snapshot_id,
                                     std::shared_ptr<DataFile> data_file,
                                     std::vector<std::shared_ptr<DataFile>> delete_files,
                                     std::shared_ptr<Expression> residual_filter)
    : change_ordinal_(change_ordinal),
      commit_snapshot_id_(commit_snapshot_id),
      data_file_(std::move(data_file)),
      delete_files_(std::move(delete_files)),
      residual_filter_(std::move(residual_filter)) {
  ICEBERG_DCHECK(data_file_ != nullptr, "Data file cannot be null for ChangelogScanTask");
}

int64_t ChangelogScanTask::size_bytes() const {
  int64_t total_size = data_file_->file_size_in_bytes;
  for (const auto& delete_file : delete_files_) {
    ICEBERG_DCHECK(delete_file->content_size_in_bytes.has_value(),
                   "Delete file content size must be available");
    total_size +=
        (delete_file->IsDeletionVector() ? delete_file->content_size_in_bytes.value()
                                         : delete_file->file_size_in_bytes);
  }
  return total_size;
}

int32_t ChangelogScanTask::files_count() const { return 1 + delete_files_.size(); }

int64_t ChangelogScanTask::estimated_row_count() const {
  return data_file_->record_count;
}

// Generic template implementation for Make
template <typename ScanType>
Result<std::unique_ptr<TableScanBuilder<ScanType>>> TableScanBuilder<ScanType>::Make(
    const Table& table) {
  ICEBERG_PRECHECK(table.metadata() != nullptr, "Table metadata cannot be null");
  ICEBERG_PRECHECK(table.io() != nullptr, "FileIO cannot be null");
  auto builder =
      std::unique_ptr<TableScanBuilder<ScanType>>(new TableScanBuilder<ScanType>(
          table.metadata(), table.io(), table.full_name(), table.reporter()));
  return builder;
}

template <typename ScanType>
TableScanBuilder<ScanType>::TableScanBuilder(
    std::shared_ptr<TableMetadata> table_metadata, std::shared_ptr<FileIO> file_io,
    std::string table_name, std::shared_ptr<MetricsReporter> metrics_reporter)
    : metadata_(std::move(table_metadata)), io_(std::move(file_io)) {
  context_.table_name = std::move(table_name);
  context_.metrics_reporter = std::move(metrics_reporter);
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::Option(std::string key,
                                                               std::string value) {
  context_.options[std::move(key)] = std::move(value);
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::Project(
    std::shared_ptr<Schema> schema) {
  context_.projected_schema = std::move(schema);
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::CaseSensitive(
    bool case_sensitive) {
  context_.case_sensitive = case_sensitive;
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::IncludeColumnStats() {
  context_.return_column_stats = true;
  context_.columns_to_keep_stats.clear();
  requested_column_stats_.reset();
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::IncludeColumnStats(
    const std::vector<std::string>& requested_columns) {
  context_.return_column_stats = true;
  requested_column_stats_ = requested_columns;

  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::Select(
    const std::vector<std::string>& column_names) {
  context_.selected_columns = column_names;
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::Filter(
    std::shared_ptr<Expression> filter) {
  context_.filter = std::move(filter);
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::IgnoreResiduals() {
  context_.ignore_residuals = true;
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::MinRowsRequested(
    int64_t num_rows) {
  context_.min_rows_requested = num_rows;
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::PlanWith(Executor& executor) {
  context_.plan_executor = std::ref(executor);
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::UseSnapshot(int64_t snapshot_id) {
  ICEBERG_BUILDER_CHECK(!context_.snapshot_id.has_value(),
                        "Cannot override snapshot, already set snapshot id={}",
                        context_.snapshot_id.value());
  ICEBERG_BUILDER_ASSIGN_OR_RETURN(std::ignore, metadata_->SnapshotById(snapshot_id));
  context_.snapshot_id = snapshot_id;
  context_.use_snapshot_schema = true;
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::UseRef(const std::string& ref) {
  if (ref == SnapshotRef::kMainBranch) {
    context_.snapshot_id.reset();
    context_.use_snapshot_schema = false;
    return *this;
  }

  ICEBERG_BUILDER_CHECK(!context_.snapshot_id.has_value(),
                        "Cannot override ref, already set snapshot id={}",
                        context_.snapshot_id.value());
  auto iter = metadata_->refs.find(ref);
  ICEBERG_BUILDER_CHECK(iter != metadata_->refs.end(), "Cannot find ref {}", ref);
  ICEBERG_BUILDER_CHECK(iter->second != nullptr, "Ref {} is null", ref);
  const int64_t snapshot_id = iter->second->snapshot_id;
  ICEBERG_BUILDER_ASSIGN_OR_RETURN(std::ignore, metadata_->SnapshotById(snapshot_id));
  context_.snapshot_id = snapshot_id;
  context_.use_snapshot_schema = (iter->second->type() == SnapshotRefType::kTag);

  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::AsOfTime(
    int64_t timestamp_millis) {
  auto time_point_ms = TimePointMsFromUnixMs(timestamp_millis);
  ICEBERG_BUILDER_ASSIGN_OR_RETURN(
      auto snapshot_id, SnapshotUtil::SnapshotIdAsOfTime(*metadata_, time_point_ms));
  return UseSnapshot(snapshot_id);
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::FromSnapshot(
    int64_t from_snapshot_id, bool inclusive)
  requires IsIncrementalScan<ScanType>
{
  if (inclusive) {
    ICEBERG_BUILDER_ASSIGN_OR_RETURN(std::ignore,
                                     metadata_->SnapshotById(from_snapshot_id));
  }
  this->context_.from_snapshot_id = from_snapshot_id;
  this->context_.from_snapshot_id_inclusive = inclusive;
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::FromSnapshot(
    const std::string& ref, bool inclusive)
  requires IsIncrementalScan<ScanType>
{
  auto iter = metadata_->refs.find(ref);
  ICEBERG_BUILDER_CHECK(iter != metadata_->refs.end(), "Cannot find ref: {}", ref);
  ICEBERG_BUILDER_CHECK(iter->second != nullptr, "Ref {} is null", ref);
  ICEBERG_BUILDER_CHECK(iter->second->type() == SnapshotRefType::kTag,
                        "Ref {} is not a tag", ref);
  return FromSnapshot(iter->second->snapshot_id, inclusive);
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::ToSnapshot(int64_t to_snapshot_id)
  requires IsIncrementalScan<ScanType>
{
  ICEBERG_BUILDER_ASSIGN_OR_RETURN(std::ignore, metadata_->SnapshotById(to_snapshot_id));
  context_.to_snapshot_id = to_snapshot_id;
  return *this;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::ToSnapshot(const std::string& ref)
  requires IsIncrementalScan<ScanType>
{
  auto iter = metadata_->refs.find(ref);
  ICEBERG_BUILDER_CHECK(iter != metadata_->refs.end(), "Cannot find ref: {}", ref);
  ICEBERG_BUILDER_CHECK(iter->second != nullptr, "Ref {} is null", ref);
  ICEBERG_BUILDER_CHECK(iter->second->type() == SnapshotRefType::kTag,
                        "Ref {} is not a tag", ref);
  return ToSnapshot(iter->second->snapshot_id);
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::UseBranch(
    const std::string& branch)
  requires IsIncrementalScan<ScanType>
{
  auto iter = metadata_->refs.find(branch);
  ICEBERG_BUILDER_CHECK(iter != metadata_->refs.end(), "Cannot find ref: {}", branch);
  ICEBERG_BUILDER_CHECK(iter->second != nullptr, "Ref {} is null", branch);
  ICEBERG_BUILDER_CHECK(iter->second->type() == SnapshotRefType::kBranch,
                        "Ref {} is not a branch", branch);
  context_.branch = branch;
  return *this;
}

template <typename ScanType>
Status TableScanBuilder<ScanType>::ResolveColumnStatsSelection() {
  if (!requested_column_stats_.has_value()) {
    return {};
  }

  context_.columns_to_keep_stats.clear();
  context_.columns_to_keep_stats.reserve(requested_column_stats_->size());

  ICEBERG_ASSIGN_OR_RAISE(auto schema_ref, ResolveSnapshotSchema());
  const auto& schema = schema_ref.get();
  for (const auto& column_name : *requested_column_stats_) {
    ICEBERG_ASSIGN_OR_RAISE(auto field, schema->FindFieldByName(column_name));
    ICEBERG_CHECK(field.has_value(), "Cannot find stats column: {}", column_name);
    context_.columns_to_keep_stats.insert(field.value().get().field_id());
  }

  return {};
}

template <typename ScanType>
Result<std::reference_wrapper<const std::shared_ptr<Schema>>>
TableScanBuilder<ScanType>::ResolveSnapshotSchema() {
  if (snapshot_schema_ == nullptr) {
    if (context_.snapshot_id.has_value() && context_.use_snapshot_schema) {
      ICEBERG_ASSIGN_OR_RAISE(auto snapshot,
                              metadata_->SnapshotById(*context_.snapshot_id));
      int32_t schema_id = snapshot->schema_id.value_or(Schema::kInitialSchemaId);
      ICEBERG_ASSIGN_OR_RAISE(snapshot_schema_, metadata_->SchemaById(schema_id));
    } else {
      ICEBERG_ASSIGN_OR_RAISE(snapshot_schema_, metadata_->Schema());
    }
  }
  ICEBERG_CHECK(snapshot_schema_ != nullptr, "Snapshot schema is null");
  return snapshot_schema_;
}

template <typename ScanType>
TableScanBuilder<ScanType>& TableScanBuilder<ScanType>::ReportWith(
    std::shared_ptr<MetricsReporter> reporter) {
  context_.metrics_reporter =
      MetricsReporters::Combine(context_.metrics_reporter, std::move(reporter));
  return *this;
}

template <typename ScanType>
Result<std::unique_ptr<ScanType>> TableScanBuilder<ScanType>::Build() {
  ICEBERG_RETURN_UNEXPECTED(CheckErrors());
  ICEBERG_RETURN_UNEXPECTED(ResolveColumnStatsSelection());
  ICEBERG_RETURN_UNEXPECTED(context_.Validate());

  ICEBERG_ASSIGN_OR_RAISE(auto schema, ResolveSnapshotSchema());
  return ScanType::Make(metadata_, schema.get(), io_, std::move(context_));
}

// Explicit template instantiations
template class ICEBERG_TEMPLATE_EXPORT TableScanBuilder<DataTableScan>;
template class ICEBERG_TEMPLATE_EXPORT TableScanBuilder<IncrementalAppendScan>;
template class ICEBERG_TEMPLATE_EXPORT TableScanBuilder<IncrementalChangelogScan>;

TableScan::TableScan(std::shared_ptr<TableMetadata> metadata,
                     std::shared_ptr<Schema> schema, std::shared_ptr<FileIO> file_io,
                     internal::TableScanContext context)
    : metadata_(std::move(metadata)),
      schema_(std::move(schema)),
      io_(std::move(file_io)),
      context_(std::move(context)) {}

TableScan::~TableScan() = default;

const std::shared_ptr<TableMetadata>& TableScan::metadata() const { return metadata_; }

Result<std::shared_ptr<Snapshot>> TableScan::snapshot() const {
  auto snapshot_id = context_.snapshot_id ? context_.snapshot_id.value()
                                          : metadata_->current_snapshot_id;
  if (snapshot_id == kInvalidSnapshotId) {
    return std::shared_ptr<Snapshot>{nullptr};
  }
  return metadata_->SnapshotById(snapshot_id);
}

Result<std::shared_ptr<Schema>> TableScan::schema() const {
  return ResolveProjectedSchema();
}

const internal::TableScanContext& TableScan::context() const { return context_; }

const std::shared_ptr<FileIO>& TableScan::io() const { return io_; }

const std::shared_ptr<Expression>& TableScan::filter() const {
  const static std::shared_ptr<Expression> true_expr = True::Instance();
  if (!context_.filter) {
    return true_expr;
  }
  return context_.filter;
}

bool TableScan::is_case_sensitive() const { return context_.case_sensitive; }

Result<std::reference_wrapper<const std::shared_ptr<Schema>>>
TableScan::ResolveProjectedSchema() const {
  if (projected_schema_ != nullptr) {
    return projected_schema_;
  }

  if (context_.selected_columns.has_value() &&
      !std::ranges::contains(context_.selected_columns.value(), Schema::kAllColumns)) {
    std::unordered_set<int32_t> required_field_ids;

    // Include columns referenced by filter
    if (context_.filter != nullptr) {
      ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(context_.filter));
      if (is_bound) {
        ICEBERG_ASSIGN_OR_RAISE(required_field_ids,
                                ReferenceVisitor::GetReferencedFieldIds(context_.filter));
      } else {
        ICEBERG_ASSIGN_OR_RAISE(auto filter, Binder::Bind(*schema_, context_.filter,
                                                          context_.case_sensitive));
        ICEBERG_ASSIGN_OR_RAISE(required_field_ids,
                                ReferenceVisitor::GetReferencedFieldIds(filter));
      }
    }

    // Include columns selected by option
    ICEBERG_ASSIGN_OR_RAISE(
        auto selected,
        schema_->Select(context_.selected_columns.value(), context_.case_sensitive));
    ICEBERG_ASSIGN_OR_RAISE(
        auto selected_field_ids,
        GetProjectedIdsVisitor::GetProjectedIds(*selected, /*include_struct_ids=*/true));
    required_field_ids.insert(std::make_move_iterator(selected_field_ids.begin()),
                              std::make_move_iterator(selected_field_ids.end()));

    ICEBERG_ASSIGN_OR_RAISE(projected_schema_, schema_->Project(required_field_ids));
  } else if (context_.projected_schema != nullptr) {
    projected_schema_ = context_.projected_schema;
  } else {
    projected_schema_ = schema_;
  }

  return projected_schema_;
}

const std::vector<std::string>& TableScan::ScanColumns() const {
  return context_.return_column_stats ? kScanColumnsWithStats : kScanColumns;
}

// Friend function template for IncrementalScan that implements the shared PlanFiles
// logic. It resolves the from/to snapshot range from the scan context and delegates
// to the two-arg virtual PlanFiles() override in the concrete subclass.
// Defined as a friend to access the protected two-arg PlanFiles().
template <typename ScanTaskType>
Result<std::vector<std::shared_ptr<ScanTaskType>>> ResolvePlanFiles(
    const IncrementalScan<ScanTaskType>& scan) {
  if (IsScanCurrentLineage(scan.context())) {
    if (scan.metadata()->current_snapshot_id == kInvalidSnapshotId) {
      return std::vector<std::shared_ptr<ScanTaskType>>{};
    }
  }

  ICEBERG_ASSIGN_OR_RAISE(
      int64_t to_snapshot_id_inclusive,
      internal::ToSnapshotIdInclusive(scan.context(), *scan.metadata()));
  ICEBERG_ASSIGN_OR_RAISE(
      std::optional<int64_t> from_snapshot_id_exclusive,
      internal::FromSnapshotIdExclusive(scan.context(), *scan.metadata(),
                                        to_snapshot_id_inclusive));

  return scan.PlanFiles(from_snapshot_id_exclusive, to_snapshot_id_inclusive);
}

template Result<std::vector<std::shared_ptr<FileScanTask>>> ResolvePlanFiles(
    const IncrementalScan<FileScanTask>& scan);
template Result<std::vector<std::shared_ptr<ChangelogScanTask>>> ResolvePlanFiles(
    const IncrementalScan<ChangelogScanTask>& scan);

}  // namespace iceberg