feat: basic table scan planning

gty404 · gty404 · commit e971cc4711d2 · 2025-05-27T14:43:17.000+08:00
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
@@ -38,6 +38,7 @@ set(ICEBERG_SOURCES
     sort_order.cc
     statistics_file.cc
     table_metadata.cc
+    table_scan.cc
     transform.cc
     transform_function.cc
     type.cc
diff --git a/src/iceberg/result.h b/src/iceberg/result.h
@@ -31,6 +31,7 @@ namespace iceberg {
 enum class ErrorKind {
   kAlreadyExists,
   kCommitStateUnknown,
+  kDataInvalid,
   kInvalidArgument,
   kInvalidExpression,
   kInvalidSchema,
@@ -65,14 +66,15 @@ using Status = Result<void>;
 /// \brief Macro to define error creation functions
 #define DEFINE_ERROR_FUNCTION(name)                                           \
   template <typename... Args>                                                 \
-  inline auto name(const std::format_string<Args...> fmt, Args&&... args)     \
-      -> unexpected<Error> {                                                  \
+  inline auto name(const std::format_string<Args...> fmt,                     \
+                   Args&&... args) -> unexpected<Error> {                     \
     return unexpected<Error>(                                                 \
         {ErrorKind::k##name, std::format(fmt, std::forward<Args>(args)...)}); \
   }
 
 DEFINE_ERROR_FUNCTION(AlreadyExists)
 DEFINE_ERROR_FUNCTION(CommitStateUnknown)
+DEFINE_ERROR_FUNCTION(DataInvalid)
 DEFINE_ERROR_FUNCTION(InvalidArgument)
 DEFINE_ERROR_FUNCTION(InvalidExpression)
 DEFINE_ERROR_FUNCTION(InvalidSchema)
diff --git a/src/iceberg/table.h b/src/iceberg/table.h
@@ -88,21 +88,20 @@ class ICEBERG_EXPORT Table {
   /// \return a vector of history entries
   virtual const std::vector<std::shared_ptr<HistoryEntry>>& history() const = 0;
 
-  /// \brief Create a new table scan for this table
+  /// \brief Create a new table scan builder for this table
   ///
-  /// Once a table scan is created, it can be refined to project columns and filter data.
-  virtual std::unique_ptr<TableScan> NewScan() const = 0;
+  /// Once a table scan builder is created, it can be refined to project columns and
+  /// filter data.
+  virtual std::unique_ptr<TableScanBuilder> NewScan() const = 0;
 
   /// \brief Create a new append API to add files to this table and commit
   virtual std::shared_ptr<AppendFiles> NewAppend() = 0;
 
   /// \brief Create a new transaction API to commit multiple table operations at once
   virtual std::unique_ptr<Transaction> NewTransaction() = 0;
 
-  /// TODO(wgtmac): design of FileIO is not finalized yet. We intend to use an
-  /// IO-less design in the core library.
   // /// \brief Returns a FileIO to read and write table data and metadata files
-  // virtual std::shared_ptr<FileIO> io() const = 0;
+  virtual std::shared_ptr<FileIO> io() const = 0;
 
   /// \brief Returns a LocationProvider to provide locations for new data files
   virtual std::unique_ptr<LocationProvider> location_provider() const = 0;
diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
@@ -0,0 +1,122 @@
+/*
+ * 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 "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/manifest_reader.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+TableScanBuilder::TableScanBuilder(const Table& table) : table_(table) {}
+
+TableScanBuilder& TableScanBuilder::WithColumnNames(
+    const std::vector<std::string>& column_names) {
+  column_names_ = column_names;
+  return *this;
+}
+
+TableScanBuilder& TableScanBuilder::WithSnapshotId(int64_t snapshot_id) {
+  snapshot_id_ = snapshot_id;
+  return *this;
+}
+
+TableScanBuilder& TableScanBuilder::WithFilter(
+    const std::shared_ptr<Expression>& filter) {
+  filter_ = filter;
+  return *this;
+}
+
+Result<std::unique_ptr<TableScan>> TableScanBuilder::Build() {
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot, snapshot_id_ ? table_.snapshot(*snapshot_id_)
+                                                      : Result<std::shared_ptr<Snapshot>>(
+                                                            table_.current_snapshot()));
+
+  auto ResolveSchema = [&]() -> Result<std::shared_ptr<Schema>> {
+    if (snapshot->schema_id) {
+      const auto& schemas = table_.schemas();
+      if (auto it = schemas.find(*snapshot->schema_id); it != schemas.end()) {
+        return it->second;
+      }
+      return DataInvalid("Schema {} in snapshot {} is not found", *snapshot->schema_id,
+                         snapshot->snapshot_id);
+    }
+    return table_.schema();
+  };
+
+  ICEBERG_ASSIGN_OR_RAISE(auto schema, ResolveSchema());
+
+  std::vector<int32_t> field_ids;
+  field_ids.reserve(column_names_.size());
+  for (const auto& column_name : column_names_) {
+    auto field_opt = schema->GetFieldByName(column_name);
+    if (!field_opt) {
+      return InvalidArgument("Column {} not found in schema", column_name);
+    }
+    field_ids.emplace_back(field_opt.value().get().field_id());
+  }
+
+  auto context = std::make_unique<TableScan::ScanContext>(
+      std::move(snapshot), std::move(schema), std::move(field_ids), std::move(filter_));
+  return std::make_unique<TableScan>(std::move(context), table_.io());
+}
+
+TableScan::TableScan(std::unique_ptr<ScanContext> context,
+                     std::shared_ptr<FileIO> file_io)
+    : context_(std::move(context)), file_io_(std::move(file_io)) {}
+
+Result<std::vector<std::unique_ptr<FileScanTask>>> TableScan::PlanFiles() const {
+  ICEBERG_ASSIGN_OR_RAISE(auto manifest_list_reader,
+                          CreateManifestListReader(context_->snapshot_->manifest_list));
+  ICEBERG_ASSIGN_OR_RAISE(auto manifest_files, manifest_list_reader->Files());
+
+  std::vector<std::unique_ptr<FileScanTask>> tasks;
+  for (const auto& manifest_file : manifest_files) {
+    ICEBERG_ASSIGN_OR_RAISE(auto manifest_reader,
+                            CreateManifestReader(manifest_file->manifest_path));
+    ICEBERG_ASSIGN_OR_RAISE(auto manifests, manifest_reader->Entries());
+
+    for (const auto& manifest : manifests) {
+      const auto& data_file = manifest->data_file;
+      tasks.emplace_back(std::make_unique<FileScanTask>(
+          data_file.file_path, 0, data_file.file_size_in_bytes, data_file.record_count,
+          data_file.content, data_file.file_format, context_->schema_,
+          context_->field_ids_, context_->filter_));
+    }
+  }
+  return tasks;
+}
+
+Result<std::unique_ptr<ManifestListReader>> TableScan::CreateManifestListReader(
+    const std::string& file_path) const {
+  return NotImplemented("manifest list reader");
+}
+
+Result<std::unique_ptr<ManifestReader>> TableScan::CreateManifestReader(
+    const std::string& file_path) const {
+  return NotImplemented("manifest reader");
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
@@ -0,0 +1,117 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <string>
+#include <vector>
+
+#include "iceberg/expression/expression.h"
+#include "iceberg/file_io.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Builder class for creating TableScan instances.
+class ICEBERG_EXPORT TableScanBuilder {
+ public:
+  /// \brief Constructs a TableScanBuilder for the given table.
+  /// \param table Reference to the table to scan.
+  explicit TableScanBuilder(const Table& table);
+
+  /// \brief Sets the snapshot ID to scan.
+  /// \param snapshot_id The ID of the snapshot.
+  /// \return Reference to the builder.
+  TableScanBuilder& WithSnapshotId(int64_t snapshot_id);
+
+  /// \brief Selects columns to include in the scan.
+  /// \param column_names A list of column names.
+  /// \return Reference to the builder.
+  TableScanBuilder& WithColumnNames(const std::vector<std::string>& column_names);
+
+  /// \brief Applies a filter expression to the scan.
+  /// \param filter Filter expression to use.
+  /// \return Reference to the builder.
+  TableScanBuilder& WithFilter(const std::shared_ptr<Expression>& filter);
+
+  /// \brief Builds and returns a TableScan instance.
+  /// \return A Result containing the TableScan or an error.
+  Result<std::unique_ptr<TableScan>> Build();
+
+ private:
+  const Table& table_;
+  std::vector<std::string> column_names_;
+  std::optional<int64_t> snapshot_id_;
+  std::shared_ptr<Expression> filter_;
+};
+
+/// \brief Represents a configured scan operation on a table.
+class ICEBERG_EXPORT TableScan {
+ public:
+  /// \brief Scan context holding snapshot and scan-specific metadata.
+  struct ScanContext {
+    std::shared_ptr<Snapshot> snapshot_;  ///< Snapshot to scan.
+    std::shared_ptr<Schema> schema_;      ///< Projected schema.
+    std::vector<int32_t> field_ids_;      ///< Field IDs of selected columns.
+    std::shared_ptr<Expression> filter_;  ///< Filter expression to apply.
+  };
+
+  /// \brief Constructs a TableScan with the given context and file I/O.
+  /// \param context Scan context including snapshot, schema, and filter.
+  /// \param file_io File I/O instance for reading manifests and data files.
+  TableScan(std::unique_ptr<ScanContext> context, std::shared_ptr<FileIO> file_io);
+
+  /// \brief Plans the scan tasks by resolving manifests and data files.
+  ///
+  /// Returns a list of file scan tasks if successful.
+  /// \return A Result containing scan tasks or an error.
+  Result<std::vector<std::unique_ptr<FileScanTask>>> PlanFiles() const;
+
+ private:
+  /// \brief Creates a reader for the manifest list.
+  /// \param file_path Path to the manifest list file.
+  /// \return A Result containing the reader or an error.
+  Result<std::unique_ptr<ManifestListReader>> CreateManifestListReader(
+      const std::string& file_path) const;
+
+  /// \brief Creates a reader for a manifest file.
+  /// \param file_path Path to the manifest file.
+  /// \return A Result containing the reader or an error.
+  Result<std::unique_ptr<ManifestReader>> CreateManifestReader(
+      const std::string& file_path) const;
+
+  std::unique_ptr<ScanContext> context_;
+  std::shared_ptr<FileIO> file_io_;
+};
+
+/// \brief Represents a task to scan a portion of a data file.
+struct ICEBERG_EXPORT FileScanTask {
+  std::string file_path_;                 ///< Path to the data file.
+  uint64_t start_;                        ///< Start byte offset.
+  uint64_t length_;                       ///< Length in bytes to scan.
+  std::optional<uint64_t> record_count_;  ///< Optional number of records.
+  DataFile::Content file_content_;        ///< Type of file content.
+  FileFormatType file_format_;            ///< Format of the data file.
+  std::shared_ptr<Schema> schema_;        ///< Projected schema.
+  std::vector<int32_t> field_ids_;        ///< Field IDs to project.
+  std::shared_ptr<Expression> filter_;    ///< Filter expression to apply.
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
@@ -91,6 +91,9 @@ class LocationProvider;
 class SortField;
 class SortOrder;
 class Table;
+class TableScan;
+struct FileScanTask;
+class TableScanBuilder;
 class Transaction;
 class Transform;
 class TransformFunction;
@@ -119,11 +122,12 @@ class MetadataUpdate;
 class UpdateRequirement;
 
 class AppendFiles;
-class TableScan;
 
 struct DataFile;
 struct ManifestEntry;
 struct ManifestFile;
+class ManifestReader;
 struct ManifestList;
+class ManifestListReader;
 
 }  // namespace iceberg
