src/iceberg/table.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.h"

#include <memory>

#include "iceberg/catalog.h"
#include "iceberg/data_table_scan.h"
#include "iceberg/incremental_append_scan.h"
#include "iceberg/incremental_changelog_scan.h"
#include "iceberg/labels.h"
#include "iceberg/location_provider.h"
#include "iceberg/partition_spec.h"
#include "iceberg/result.h"
#include "iceberg/schema.h"
#include "iceberg/sort_order.h"
#include "iceberg/table_metadata.h"
#include "iceberg/table_properties.h"
#include "iceberg/table_scan.h"
#include "iceberg/transaction.h"
#include "iceberg/update/delete_files.h"
#include "iceberg/update/expire_snapshots.h"
#include "iceberg/update/fast_append.h"
#include "iceberg/update/merge_append.h"
#include "iceberg/update/overwrite_files.h"
#include "iceberg/update/replace_partitions.h"
#include "iceberg/update/rewrite_files.h"
#include "iceberg/update/row_delta.h"
#include "iceberg/update/set_snapshot.h"
#include "iceberg/update/snapshot_manager.h"
#include "iceberg/update/update_location.h"
#include "iceberg/update/update_partition_spec.h"
#include "iceberg/update/update_partition_statistics.h"
#include "iceberg/update/update_properties.h"
#include "iceberg/update/update_schema.h"
#include "iceberg/update/update_snapshot_reference.h"
#include "iceberg/update/update_sort_order.h"
#include "iceberg/update/update_statistics.h"
#include "iceberg/util/macros.h"

namespace iceberg {

Result<std::shared_ptr<Table>> Table::Make(
    TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
    std::string metadata_location, std::shared_ptr<FileIO> io,
    std::shared_ptr<Catalog> catalog, std::string full_name,
    std::shared_ptr<MetricsReporter> reporter, std::unique_ptr<Labels> labels) {
  if (metadata == nullptr) [[unlikely]] {
    return InvalidArgument("Metadata cannot be null");
  }
  if (metadata_location.empty()) [[unlikely]] {
    return InvalidArgument("Metadata location cannot be empty");
  }
  if (io == nullptr) [[unlikely]] {
    return InvalidArgument("FileIO cannot be null");
  }
  if (catalog == nullptr) [[unlikely]] {
    return InvalidArgument("Catalog cannot be null");
  }
  return std::shared_ptr<Table>(new Table(std::move(identifier), std::move(metadata),
                                          std::move(metadata_location), std::move(io),
                                          std::move(catalog), std::move(full_name),
                                          std::move(reporter), std::move(labels)));
}

Table::~Table() = default;

Table::Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
             std::string metadata_location, std::shared_ptr<FileIO> io,
             std::shared_ptr<Catalog> catalog, std::string full_name,
             std::shared_ptr<MetricsReporter> reporter, std::unique_ptr<Labels> labels)
    : identifier_(std::move(identifier)),
      full_name_(full_name.empty() ? identifier_.ToString() : std::move(full_name)),
      metadata_(std::move(metadata)),
      metadata_location_(std::move(metadata_location)),
      io_(std::move(io)),
      catalog_(std::move(catalog)),
      reporter_(std::move(reporter)),
      metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())),
      labels_(labels && !labels->empty() ? std::move(labels) : nullptr) {}

const std::string& Table::uuid() const { return metadata_->table_uuid; }

Status Table::Refresh() {
  ICEBERG_ASSIGN_OR_RAISE(auto refreshed_table, catalog_->LoadTable(identifier_));
  if (metadata_location_ != refreshed_table->metadata_file_location()) {
    metadata_ = std::move(refreshed_table->metadata_);
    metadata_location_ = std::string(refreshed_table->metadata_file_location());
    io_ = std::move(refreshed_table->io_);
    metadata_cache_ = std::make_unique<TableMetadataCache>(metadata_.get());
  }
  return {};
}

Result<std::shared_ptr<Schema>> Table::schema() const { return metadata_->Schema(); }

Result<std::reference_wrapper<const std::unordered_map<int32_t, std::shared_ptr<Schema>>>>
Table::schemas() const {
  return metadata_cache_->GetSchemasById();
}

Result<std::shared_ptr<PartitionSpec>> Table::spec() const {
  return metadata_->PartitionSpec();
}

Result<std::reference_wrapper<
    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>>>
Table::specs() const {
  return metadata_cache_->GetPartitionSpecsById();
}

Result<std::shared_ptr<SortOrder>> Table::sort_order() const {
  return metadata_->SortOrder();
}

Result<
    std::reference_wrapper<const std::unordered_map<int32_t, std::shared_ptr<SortOrder>>>>
Table::sort_orders() const {
  return metadata_cache_->GetSortOrdersById();
}

const TableProperties& Table::properties() const { return metadata_->properties; }

std::string_view Table::metadata_file_location() const { return metadata_location_; }

std::string_view Table::location() const { return metadata_->location; }

TimePointMs Table::last_updated_ms() const { return metadata_->last_updated_ms; }

Result<std::shared_ptr<Snapshot>> Table::current_snapshot() const {
  return metadata_->Snapshot();
}

Result<std::shared_ptr<Snapshot>> Table::SnapshotById(int64_t snapshot_id) const {
  return metadata_->SnapshotById(snapshot_id);
}

const std::vector<std::shared_ptr<Snapshot>>& Table::snapshots() const {
  return metadata_->snapshots;
}

const std::vector<SnapshotLogEntry>& Table::history() const {
  return metadata_->snapshot_log;
}

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

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

const std::shared_ptr<Catalog>& Table::catalog() const { return catalog_; }

const std::shared_ptr<MetricsReporter>& Table::reporter() const { return reporter_; }

const Labels& Table::labels() const {
  if (labels_) {
    return *labels_;
  }
  static const Labels kEmpty;
  return kEmpty;
}

Result<std::unique_ptr<LocationProvider>> Table::location_provider() const {
  return LocationProvider::Make(metadata_->location, metadata_->properties);
}

Result<std::unique_ptr<DataTableScanBuilder>> Table::NewScan() const {
  return DataTableScanBuilder::Make(*this);
}

Result<std::unique_ptr<IncrementalAppendScanBuilder>> Table::NewIncrementalAppendScan()
    const {
  return IncrementalAppendScanBuilder::Make(*this);
}

Result<std::unique_ptr<IncrementalChangelogScanBuilder>>
Table::NewIncrementalChangelogScan() const {
  return IncrementalChangelogScanBuilder::Make(*this);
}

Result<std::shared_ptr<Transaction>> Table::NewTransaction() {
  // Create a brand new transaction object for the table. Users are expected to commit the
  // transaction manually.
  return Transaction::Make(shared_from_this(), TransactionKind::kUpdate);
}

Result<std::shared_ptr<UpdatePartitionSpec>> Table::NewUpdatePartitionSpec() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return UpdatePartitionSpec::Make(std::move(ctx));
}

Result<std::shared_ptr<UpdateProperties>> Table::NewUpdateProperties() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return UpdateProperties::Make(std::move(ctx));
}

Result<std::shared_ptr<UpdateSortOrder>> Table::NewUpdateSortOrder() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return UpdateSortOrder::Make(std::move(ctx));
}

Result<std::shared_ptr<UpdateSchema>> Table::NewUpdateSchema() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return UpdateSchema::Make(std::move(ctx));
}

Result<std::shared_ptr<ExpireSnapshots>> Table::NewExpireSnapshots() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return ExpireSnapshots::Make(std::move(ctx));
}

Result<std::shared_ptr<UpdateLocation>> Table::NewUpdateLocation() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return UpdateLocation::Make(std::move(ctx));
}

Result<std::shared_ptr<FastAppend>> Table::NewFastAppend() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return FastAppend::Make(name().name, std::move(ctx));
}

Result<std::shared_ptr<MergeAppend>> Table::NewMergeAppend() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return MergeAppend::Make(name().name, std::move(ctx));
}

Result<std::shared_ptr<DeleteFiles>> Table::NewDeleteFiles() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return DeleteFiles::Make(name().name, std::move(ctx));
}

Result<std::shared_ptr<RowDelta>> Table::NewRowDelta() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return RowDelta::Make(name().name, std::move(ctx));
}

Result<std::shared_ptr<OverwriteFiles>> Table::NewOverwrite() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return OverwriteFiles::Make(name().name, std::move(ctx));
}

Result<std::shared_ptr<RewriteFiles>> Table::NewRewriteFiles() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return RewriteFiles::Make(name().name, std::move(ctx));
}

Result<std::shared_ptr<ReplacePartitions>> Table::NewReplacePartitions() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return ReplacePartitions::Make(name().name, std::move(ctx));
}

Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return UpdateStatistics::Make(std::move(ctx));
}

Result<std::shared_ptr<UpdatePartitionStatistics>> Table::NewUpdatePartitionStatistics() {
  ICEBERG_ASSIGN_OR_RAISE(
      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
  return UpdatePartitionStatistics::Make(std::move(ctx));
}

Result<std::shared_ptr<SnapshotManager>> Table::NewSnapshotManager() {
  return SnapshotManager::Make(shared_from_this());
}

Result<std::shared_ptr<StagedTable>> StagedTable::Make(
    TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
    std::string metadata_location, std::shared_ptr<FileIO> io,
    std::shared_ptr<Catalog> catalog, std::string full_name,
    std::shared_ptr<MetricsReporter> reporter) {
  if (metadata == nullptr) [[unlikely]] {
    return InvalidArgument("Metadata cannot be null");
  }
  if (io == nullptr) [[unlikely]] {
    return InvalidArgument("FileIO cannot be null");
  }
  if (catalog == nullptr) [[unlikely]] {
    return InvalidArgument("Catalog cannot be null");
  }
  return std::shared_ptr<StagedTable>(
      new StagedTable(std::move(identifier), std::move(metadata),
                      std::move(metadata_location), std::move(io), std::move(catalog),
                      std::move(full_name), std::move(reporter), /*labels=*/nullptr));
}

StagedTable::~StagedTable() = default;

Result<std::unique_ptr<DataTableScanBuilder>> StagedTable::NewScan() const {
  return NotSupported("Cannot scan a staged table");
}

Result<std::shared_ptr<StaticTable>> StaticTable::Make(
    TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
    std::string metadata_location, std::shared_ptr<FileIO> io, std::string full_name) {
  if (metadata == nullptr) [[unlikely]] {
    return InvalidArgument("Metadata cannot be null");
  }
  if (io == nullptr) [[unlikely]] {
    return InvalidArgument("FileIO cannot be null");
  }
  return std::shared_ptr<StaticTable>(
      new StaticTable(std::move(identifier), std::move(metadata),
                      std::move(metadata_location), std::move(io), /*catalog=*/nullptr,
                      std::move(full_name), /*reporter=*/nullptr, /*labels=*/nullptr));
}

StaticTable::~StaticTable() = default;

Status StaticTable::Refresh() { return NotSupported("Cannot refresh a static table"); }

Result<std::shared_ptr<Transaction>> StaticTable::NewTransaction() {
  return NotSupported("Cannot create a transaction for a static table");
}

Result<std::shared_ptr<UpdateProperties>> StaticTable::NewUpdateProperties() {
  return NotSupported("Cannot create an update properties for a static table");
}

Result<std::shared_ptr<UpdateSchema>> StaticTable::NewUpdateSchema() {
  return NotSupported("Cannot create an update schema for a static table");
}

Result<std::shared_ptr<UpdateLocation>> StaticTable::NewUpdateLocation() {
  return NotSupported("Cannot create an update location for a static table");
}

Result<std::shared_ptr<UpdatePartitionSpec>> StaticTable::NewUpdatePartitionSpec() {
  return NotSupported("Cannot create an update partition spec for a static table");
}

Result<std::shared_ptr<UpdateSortOrder>> StaticTable::NewUpdateSortOrder() {
  return NotSupported("Cannot create an update sort order for a static table");
}

Result<std::shared_ptr<ExpireSnapshots>> StaticTable::NewExpireSnapshots() {
  return NotSupported("Cannot create an expire snapshots for a static table");
}

Result<std::shared_ptr<UpdateStatistics>> StaticTable::NewUpdateStatistics() {
  return NotSupported("Cannot create an update statistics for a static table");
}

Result<std::shared_ptr<UpdatePartitionStatistics>>
StaticTable::NewUpdatePartitionStatistics() {
  return NotSupported("Cannot create an update partition statistics for a static table");
}

Result<std::shared_ptr<FastAppend>> StaticTable::NewFastAppend() {
  return NotSupported("Cannot create a fast append for a static table");
}

Result<std::shared_ptr<MergeAppend>> StaticTable::NewMergeAppend() {
  return NotSupported("Cannot create a merge append for a static table");
}

Result<std::shared_ptr<DeleteFiles>> StaticTable::NewDeleteFiles() {
  return NotSupported("Cannot create delete files for a static table");
}

Result<std::shared_ptr<RowDelta>> StaticTable::NewRowDelta() {
  return NotSupported("Cannot create a row delta for a static table");
}

Result<std::shared_ptr<OverwriteFiles>> StaticTable::NewOverwrite() {
  return NotSupported("Cannot create an overwrite for a static table");
}

Result<std::shared_ptr<RewriteFiles>> StaticTable::NewRewriteFiles() {
  return NotSupported("Cannot create a rewrite files for a static table");
}

Result<std::shared_ptr<ReplacePartitions>> StaticTable::NewReplacePartitions() {
  return NotSupported("Cannot replace partitions for a static table");
}

Result<std::shared_ptr<SnapshotManager>> StaticTable::NewSnapshotManager() {
  return NotSupported("Cannot create a snapshot manager for a static table");
}

}  // namespace iceberg