Merge branch 'main' into table_location_provider

Author: wgtmac

src/iceberg/CMakeLists.txt

@@ -82,6 +82,7 @@ set(ICEBERG_SOURCES
     transform_function.cc
     type.cc
     update/pending_update.cc
+    update/snapshot_update.cc
     update/update_partition_spec.cc
     update/update_properties.cc
     update/update_schema.cc

src/iceberg/avro/avro_schema_util.cc

@@ -17,7 +17,6 @@
  * under the License.
  */
 
-#include <charconv>
 #include <format>
 #include <mutex>
 #include <sstream>
@@ -40,6 +39,7 @@
 #include "iceberg/schema_util_internal.h"
 #include "iceberg/util/formatter.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
 #include "iceberg/util/visit_type.h"
 
 namespace iceberg::avro {
@@ -471,13 +471,7 @@ Result<int32_t> GetId(const ::avro::NodePtr& node, const std::string& attr_name,
     return InvalidSchema("Missing avro attribute: {}", attr_name);
   }
 
-  int32_t id;
-  const auto& id_value = id_str.value();
-  auto [_, ec] = std::from_chars(id_value.data(), id_value.data() + id_value.size(), id);
-  if (ec != std::errc()) {
-    return InvalidSchema("Invalid {}: {}", attr_name, id_value);
-  }
-  return id;
+  return StringUtils::ParseInt<int32_t>(id_str.value());
 }
 
 Result<int32_t> GetElementId(const ::avro::NodePtr& node) {

src/iceberg/expression/binder.cc

@@ -19,6 +19,9 @@
 
 #include "iceberg/expression/binder.h"
 
+#include "iceberg/result.h"
+#include "iceberg/util/macros.h"
+
 namespace iceberg {
 
 Binder::Binder(const Schema& schema, bool case_sensitive)
@@ -54,30 +57,30 @@ Result<std::shared_ptr<Expression>> Binder::Or(
 
 Result<std::shared_ptr<Expression>> Binder::Predicate(
     const std::shared_ptr<UnboundPredicate>& pred) {
-  ICEBERG_DCHECK(pred != nullptr, "Predicate cannot be null");
+  ICEBERG_PRECHECK(pred != nullptr, "Predicate cannot be null");
   return pred->Bind(schema_, case_sensitive_);
 }
 
 Result<std::shared_ptr<Expression>> Binder::Predicate(
     const std::shared_ptr<BoundPredicate>& pred) {
-  ICEBERG_DCHECK(pred != nullptr, "Predicate cannot be null");
+  ICEBERG_PRECHECK(pred != nullptr, "Predicate cannot be null");
   return InvalidExpression("Found already bound predicate: {}", pred->ToString());
 }
 
 Result<std::shared_ptr<Expression>> Binder::Aggregate(
     const std::shared_ptr<BoundAggregate>& aggregate) {
-  ICEBERG_DCHECK(aggregate != nullptr, "Aggregate cannot be null");
+  ICEBERG_PRECHECK(aggregate != nullptr, "Aggregate cannot be null");
   return InvalidExpression("Found already bound aggregate: {}", aggregate->ToString());
 }
 
 Result<std::shared_ptr<Expression>> Binder::Aggregate(
     const std::shared_ptr<UnboundAggregate>& aggregate) {
-  ICEBERG_DCHECK(aggregate != nullptr, "Aggregate cannot be null");
+  ICEBERG_PRECHECK(aggregate != nullptr, "Aggregate cannot be null");
   return aggregate->Bind(schema_, case_sensitive_);
 }
 
 Result<bool> IsBoundVisitor::IsBound(const std::shared_ptr<Expression>& expr) {
-  ICEBERG_DCHECK(expr != nullptr, "Expression cannot be null");
+  ICEBERG_PRECHECK(expr != nullptr, "Expression cannot be null");
   IsBoundVisitor visitor;
   return Visit<bool, IsBoundVisitor>(expr, visitor);
 }
@@ -113,4 +116,54 @@ Result<bool> IsBoundVisitor::Aggregate(
   return false;
 }
 
+Result<std::unordered_set<int32_t>> ReferenceVisitor::GetReferencedFieldIds(
+    const std::shared_ptr<Expression>& expr) {
+  ICEBERG_PRECHECK(expr != nullptr, "Expression cannot be null");
+  ReferenceVisitor visitor;
+  return Visit<FieldIdsSetRef, ReferenceVisitor>(expr, visitor);
+}
+
+Result<FieldIdsSetRef> ReferenceVisitor::AlwaysTrue() { return referenced_field_ids_; }
+
+Result<FieldIdsSetRef> ReferenceVisitor::AlwaysFalse() { return referenced_field_ids_; }
+
+Result<FieldIdsSetRef> ReferenceVisitor::Not(
+    [[maybe_unused]] const FieldIdsSetRef& child_result) {
+  return referenced_field_ids_;
+}
+
+Result<FieldIdsSetRef> ReferenceVisitor::And(
+    [[maybe_unused]] const FieldIdsSetRef& left_result,
+    [[maybe_unused]] const FieldIdsSetRef& right_result) {
+  return referenced_field_ids_;
+}
+
+Result<FieldIdsSetRef> ReferenceVisitor::Or(
+    [[maybe_unused]] const FieldIdsSetRef& left_result,
+    [[maybe_unused]] const FieldIdsSetRef& right_result) {
+  return referenced_field_ids_;
+}
+
+Result<FieldIdsSetRef> ReferenceVisitor::Predicate(
+    const std::shared_ptr<BoundPredicate>& pred) {
+  referenced_field_ids_.insert(pred->reference()->field_id());
+  return referenced_field_ids_;
+}
+
+Result<FieldIdsSetRef> ReferenceVisitor::Predicate(
+    [[maybe_unused]] const std::shared_ptr<UnboundPredicate>& pred) {
+  return InvalidExpression("Cannot get referenced field IDs from unbound predicate");
+}
+
+Result<FieldIdsSetRef> ReferenceVisitor::Aggregate(
+    const std::shared_ptr<BoundAggregate>& aggregate) {
+  referenced_field_ids_.insert(aggregate->reference()->field_id());
+  return referenced_field_ids_;
+}
+
+Result<FieldIdsSetRef> ReferenceVisitor::Aggregate(
+    [[maybe_unused]] const std::shared_ptr<UnboundAggregate>& aggregate) {
+  return InvalidExpression("Cannot get referenced field IDs from unbound aggregate");
+}
+
 }  // namespace iceberg

src/iceberg/expression/binder.h

@@ -22,6 +22,9 @@
 /// \file iceberg/expression/binder.h
 /// Bind an expression to a schema.
 
+#include <functional>
+#include <unordered_set>
+
 #include "iceberg/expression/expression_visitor.h"
 
 namespace iceberg {
@@ -73,6 +76,31 @@ class ICEBERG_EXPORT IsBoundVisitor : public ExpressionVisitor<bool> {
   Result<bool> Aggregate(const std::shared_ptr<UnboundAggregate>& aggregate) override;
 };
 
-// TODO(gangwu): add the Java parity `ReferenceVisitor`
+using FieldIdsSetRef = std::reference_wrapper<std::unordered_set<int32_t>>;
+
+/// \brief Visitor to collect referenced field IDs from an expression.
+class ICEBERG_EXPORT ReferenceVisitor : public ExpressionVisitor<FieldIdsSetRef> {
+ public:
+  static Result<std::unordered_set<int32_t>> GetReferencedFieldIds(
+      const std::shared_ptr<Expression>& expr);
+
+  Result<FieldIdsSetRef> AlwaysTrue() override;
+  Result<FieldIdsSetRef> AlwaysFalse() override;
+  Result<FieldIdsSetRef> Not(const FieldIdsSetRef& child_result) override;
+  Result<FieldIdsSetRef> And(const FieldIdsSetRef& left_result,
+                             const FieldIdsSetRef& right_result) override;
+  Result<FieldIdsSetRef> Or(const FieldIdsSetRef& left_result,
+                            const FieldIdsSetRef& right_result) override;
+  Result<FieldIdsSetRef> Predicate(const std::shared_ptr<BoundPredicate>& pred) override;
+  Result<FieldIdsSetRef> Predicate(
+      const std::shared_ptr<UnboundPredicate>& pred) override;
+  Result<FieldIdsSetRef> Aggregate(
+      const std::shared_ptr<BoundAggregate>& aggregate) override;
+  Result<FieldIdsSetRef> Aggregate(
+      const std::shared_ptr<UnboundAggregate>& aggregate) override;
+
+ private:
+  std::unordered_set<int32_t> referenced_field_ids_;
+};
 
 }  // namespace iceberg

src/iceberg/json_internal.cc

@@ -399,7 +399,7 @@ nlohmann::json ToJson(const Snapshot& snapshot) {
   json[kTimestampMs] = UnixMsFromTimePointMs(snapshot.timestamp_ms);
   json[kManifestList] = snapshot.manifest_list;
   // If there is an operation, write the summary map
-  if (snapshot.operation().has_value()) {
+  if (snapshot.Operation().has_value()) {
     json[kSummary] = snapshot.summary;
   }
   SetOptionalField(json, kSchemaId, snapshot.schema_id);
@@ -645,9 +645,8 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId));
   ICEBERG_ASSIGN_OR_RAISE(auto sequence_number,
                           GetJsonValueOptional<int64_t>(json, kSequenceNumber));
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto timestamp_ms,
-      GetJsonValue<int64_t>(json, kTimestampMs).and_then(TimePointMsFromUnixMs));
+  ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue<int64_t>(json, kTimestampMs));
+  auto timestamp_ms = TimePointMsFromUnixMs(unix_ms);
   ICEBERG_ASSIGN_OR_RAISE(auto manifest_list,
                           GetJsonValue<std::string>(json, kManifestList));
 
@@ -781,9 +780,8 @@ nlohmann::json ToJson(const SnapshotLogEntry& snapshot_log_entry) {
 
 Result<SnapshotLogEntry> SnapshotLogEntryFromJson(const nlohmann::json& json) {
   SnapshotLogEntry snapshot_log_entry;
-  ICEBERG_ASSIGN_OR_RAISE(
-      snapshot_log_entry.timestamp_ms,
-      GetJsonValue<int64_t>(json, kTimestampMs).and_then(TimePointMsFromUnixMs));
+  ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue<int64_t>(json, kTimestampMs));
+  snapshot_log_entry.timestamp_ms = TimePointMsFromUnixMs(unix_ms);
   ICEBERG_ASSIGN_OR_RAISE(snapshot_log_entry.snapshot_id,
                           GetJsonValue<int64_t>(json, kSnapshotId));
   return snapshot_log_entry;
@@ -798,9 +796,8 @@ nlohmann::json ToJson(const MetadataLogEntry& metadata_log_entry) {
 
 Result<MetadataLogEntry> MetadataLogEntryFromJson(const nlohmann::json& json) {
   MetadataLogEntry metadata_log_entry;
-  ICEBERG_ASSIGN_OR_RAISE(
-      metadata_log_entry.timestamp_ms,
-      GetJsonValue<int64_t>(json, kTimestampMs).and_then(TimePointMsFromUnixMs));
+  ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue<int64_t>(json, kTimestampMs));
+  metadata_log_entry.timestamp_ms = TimePointMsFromUnixMs(unix_ms);
   ICEBERG_ASSIGN_OR_RAISE(metadata_log_entry.metadata_file,
                           GetJsonValue<std::string>(json, kMetadataFile));
   return metadata_log_entry;
@@ -1553,9 +1550,17 @@ Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& j
                             GetJsonValueOptional<int64_t>(json, kMaxSnapshotAgeMs));
     ICEBERG_ASSIGN_OR_RAISE(auto max_ref_age,
                             GetJsonValueOptional<int64_t>(json, kMaxRefAgeMs));
-    return std::make_unique<table::SetSnapshotRef>(std::move(ref_name), snapshot_id, type,
-                                                   min_snapshots, max_snapshot_age,
-                                                   max_ref_age);
+    if (type == SnapshotRefType::kTag) {
+      ICEBERG_ASSIGN_OR_RAISE(auto tag, SnapshotRef::MakeTag(snapshot_id, max_ref_age));
+      return std::make_unique<table::SetSnapshotRef>(std::move(ref_name), *tag);
+    } else {
+      ICEBERG_CHECK(type == SnapshotRefType::kBranch,
+                    "Expected branch type for snapshot ref");
+      ICEBERG_ASSIGN_OR_RAISE(auto branch,
+                              SnapshotRef::MakeBranch(snapshot_id, min_snapshots,
+                                                      max_snapshot_age, max_ref_age));
+      return std::make_unique<table::SetSnapshotRef>(std::move(ref_name), *branch);
+    }
   }
   if (action == kActionSetProperties) {
     using StringMap = std::unordered_map<std::string, std::string>;

src/iceberg/manifest/manifest_writer.cc

@@ -369,23 +369,18 @@ Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeWriter(
     int8_t format_version, std::optional<int64_t> snapshot_id,
     std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
     std::shared_ptr<PartitionSpec> partition_spec, std::shared_ptr<Schema> current_schema,
-    std::optional<ManifestContent> content, std::optional<int64_t> first_row_id) {
+    ManifestContent content, std::optional<int64_t> first_row_id) {
   switch (format_version) {
     case 1:
       return MakeV1Writer(snapshot_id, manifest_location, std::move(file_io),
                           std::move(partition_spec), std::move(current_schema));
     case 2:
-      ICEBERG_PRECHECK(content.has_value(),
-                       "ManifestContent is required for format version 2");
       return MakeV2Writer(snapshot_id, manifest_location, std::move(file_io),
-                          std::move(partition_spec), std::move(current_schema),
-                          content.value());
+                          std::move(partition_spec), std::move(current_schema), content);
     case 3:
-      ICEBERG_PRECHECK(content.has_value(),
-                       "ManifestContent is required for format version 3");
       return MakeV3Writer(snapshot_id, first_row_id, manifest_location,
                           std::move(file_io), std::move(partition_spec),
-                          std::move(current_schema), content.value());
+                          std::move(current_schema), content);
     default:
       return NotSupported("Format version {} is not supported", format_version);
   }

src/iceberg/manifest/manifest_writer.h

@@ -28,6 +28,7 @@
 
 #include "iceberg/file_writer.h"
 #include "iceberg/iceberg_export.h"
+#include "iceberg/manifest/manifest_list.h"
 #include "iceberg/metrics.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -175,7 +176,7 @@ class ICEBERG_EXPORT ManifestWriter {
       std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
       std::shared_ptr<PartitionSpec> partition_spec,
       std::shared_ptr<Schema> current_schema,
-      std::optional<ManifestContent> content = std::nullopt,
+      ManifestContent content = ManifestContent::kData,
       std::optional<int64_t> first_row_id = std::nullopt);
 
  private:

src/iceberg/meson.build

@@ -103,6 +103,7 @@ iceberg_sources = files(
     'transform_function.cc',
     'type.cc',
     'update/pending_update.cc',
+    'update/snapshot_update.cc',
     'update/update_partition_spec.cc',
     'update/update_properties.cc',
     'update/update_schema.cc',