chore: make RowView shared_ptr

bindings/cpp/include/fluss.hpp

@@ -617,15 +617,15 @@ class GenericRow {
 
 /// Read-only row view for scan results. Zero-copy access to string and bytes data.
 ///
-/// WARNING: RowView borrows from ScanRecords. It must not outlive the ScanRecords
-/// that produced it (similar to std::string_view borrowing from std::string).
+/// RowView shares ownership of the underlying scan data via reference counting,
+/// so it can safely outlive the ScanRecords that produced it.
 class RowView : public detail::NamedGetters<RowView> {
     friend struct detail::NamedGetters<RowView>;
 
    public:
-    RowView(const ffi::ScanResultInner* inner, size_t record_idx,
-            const detail::ColumnMap* column_map)
-        : inner_(inner), record_idx_(record_idx), column_map_(column_map) {}
+    RowView(std::shared_ptr<const ffi::ScanResultInner> inner, size_t record_idx,
+            std::shared_ptr<const detail::ColumnMap> column_map)
+        : inner_(std::move(inner)), record_idx_(record_idx), column_map_(std::move(column_map)) {}
 
     // ── Index-based getters ──────────────────────────────────────────
     size_t FieldCount() const;
@@ -665,15 +665,15 @@ class RowView : public detail::NamedGetters<RowView> {
         }
         return detail::ResolveColumn(*column_map_, name);
     }
-    const ffi::ScanResultInner* inner_;
+    std::shared_ptr<const ffi::ScanResultInner> inner_;
     size_t record_idx_;
-    const detail::ColumnMap* column_map_;  // borrowed from ScanRecords (same lifetime as inner_)
+    std::shared_ptr<const detail::ColumnMap> column_map_;
 };
 
 /// A single scan record. Contains metadata and a RowView for field access.
 ///
-/// WARNING: ScanRecord contains a RowView that borrows from ScanRecords.
-/// It must not outlive the ScanRecords that produced it.
+/// ScanRecord is a value type that can be freely copied, stored, and
+/// accumulated across multiple Poll() calls.
 struct ScanRecord {
     int32_t bucket_id;
     std::optional<int64_t> partition_id;
@@ -685,13 +685,13 @@ struct ScanRecord {
 
 class ScanRecords {
    public:
-    ScanRecords() noexcept;
-    ~ScanRecords() noexcept;
+    ScanRecords() noexcept = default;
+    ~ScanRecords() noexcept = default;
 
     ScanRecords(const ScanRecords&) = delete;
     ScanRecords& operator=(const ScanRecords&) = delete;
-    ScanRecords(ScanRecords&& other) noexcept;
-    ScanRecords& operator=(ScanRecords&& other) noexcept;
+    ScanRecords(ScanRecords&&) noexcept = default;
+    ScanRecords& operator=(ScanRecords&&) noexcept = default;
 
     size_t Size() const;
     bool Empty() const;
@@ -720,9 +720,8 @@ class ScanRecords {
     /// Returns the column name-to-index map (lazy-built, cached).
     const std::shared_ptr<detail::ColumnMap>& GetColumnMap() const;
     friend class LogScanner;
-    void Destroy() noexcept;
     void BuildColumnMap() const;
-    ffi::ScanResultInner* inner_{nullptr};
+    std::shared_ptr<ffi::ScanResultInner> inner_;
     mutable std::shared_ptr<detail::ColumnMap> column_map_;
 };
 

bindings/cpp/src/table.cpp

@@ -266,32 +266,7 @@ std::string RowView::GetDecimalString(size_t idx) const {
 // ScanRecords — backed by opaque Rust ScanResultInner
 // ============================================================================
 
-ScanRecords::ScanRecords() noexcept = default;
-
-ScanRecords::~ScanRecords() noexcept { Destroy(); }
-
-void ScanRecords::Destroy() noexcept {
-    if (inner_) {
-        rust::Box<ffi::ScanResultInner>::from_raw(inner_);
-        inner_ = nullptr;
-        column_map_.reset();
-    }
-}
-
-ScanRecords::ScanRecords(ScanRecords&& other) noexcept
-    : inner_(other.inner_), column_map_(std::move(other.column_map_)) {
-    other.inner_ = nullptr;
-}
-
-ScanRecords& ScanRecords::operator=(ScanRecords&& other) noexcept {
-    if (this != &other) {
-        Destroy();
-        inner_ = other.inner_;
-        column_map_ = std::move(other.column_map_);
-        other.inner_ = nullptr;
-    }
-    return *this;
-}
+// ScanRecords constructor, destructor, move operations are all defaulted in the header.
 
 size_t ScanRecords::Size() const { return inner_ ? inner_->sv_record_count() : 0; }
 
@@ -331,7 +306,7 @@ ScanRecord ScanRecords::operator[](size_t idx) const {
                       inner_->sv_offset(idx),
                       inner_->sv_timestamp(idx),
                       static_cast<ChangeType>(inner_->sv_change_type(idx)),
-                      RowView(inner_, idx, GetColumnMap().get())};
+                      RowView(inner_, idx, GetColumnMap())};
 }
 
 ScanRecord ScanRecords::Iterator::operator*() const { return owner_->operator[](idx_); }
@@ -1107,8 +1082,10 @@ Result LogScanner::Poll(int64_t timeout_ms, ScanRecords& out) {
                                  std::string(result_box->sv_error_message()));
     }
 
-    out.Destroy();
-    out.inner_ = result_box.into_raw();
+    out.column_map_.reset();
+    out.inner_ = std::shared_ptr<ffi::ScanResultInner>(
+        result_box.into_raw(),
+        [](ffi::ScanResultInner* p) { rust::Box<ffi::ScanResultInner>::from_raw(p); });
     return utils::make_ok();
 }
 

website/docs/user-guide/cpp/api-reference.md

@@ -203,10 +203,10 @@ When using `table.NewRow()`, the `Set()` method auto-routes to the correct type
 
 ## `RowView`
 
-Read-only row view for scan results. Provides zero-copy access to string and bytes data.
+Read-only row view for scan results. Provides zero-copy access to string and bytes data. `RowView` shares ownership of the underlying scan data via reference counting, so it can safely outlive the `ScanRecords` that produced it.
 
-:::warning Lifetime
-`RowView` borrows from `ScanRecords`. It must not outlive the `ScanRecords` that produced it (similar to `std::string_view` borrowing from `std::string`).
+:::note string_view Lifetime
+`GetString()` returns `std::string_view` that borrows from the underlying data. The `string_view` is valid as long as any `RowView` (or `ScanRecord`) referencing the same poll result is alive. Copy to `std::string` if you need the value after all references are gone.
 :::
 
 ### Index-Based Getters
@@ -248,9 +248,7 @@ Read-only row view for scan results. Provides zero-copy access to string and byt
 
 ## `ScanRecord`
 
-:::warning Lifetime
-`ScanRecord` contains a `RowView` that borrows from `ScanRecords`. It must not outlive the `ScanRecords` that produced it.
-:::
+`ScanRecord` is a value type that can be freely copied, stored, and accumulated across multiple `Poll()` calls. It shares ownership of the underlying scan data via reference counting.
 
 | Field          | Type                    |  Description                     |
 |----------------|-------------------------|----------------------------------|

website/docs/user-guide/cpp/data-types.md

@@ -58,28 +58,31 @@ row.Set("nickname", nullptr);    // set to null
 
 Field values are read through `RowView` (from scan results) and `LookupResult` (from lookups), not through `GenericRow`. Both provide the same getter interface with zero-copy access to string and bytes data.
 
-:::warning Lifetime
-`RowView` borrows from `ScanRecords`. It must not outlive the `ScanRecords` that produced it (similar to `std::string_view` borrowing from `std::string`).
+`ScanRecord` is a value type — it can be freely copied, stored, and accumulated across multiple `Poll()` calls via reference counting.
+
+:::note string_view Lifetime
+`GetString()` returns `std::string_view` that borrows from the underlying data. The `string_view` is valid as long as any `ScanRecord` referencing the same poll result is alive. Copy to `std::string` if you need the value after all records are gone.
 :::
 
 ```cpp
-// DON'T — string_view dangles after ScanRecords is destroyed:
-std::string_view dangling;
-{
-    fluss::ScanRecords records;
-    scanner.Poll(5000, records);
-    dangling = records[0].row.GetString(0); // points into ScanRecords memory
-}
-// dangling is undefined behavior here — ScanRecords is gone!
-
-// DO — use values within ScanRecords lifetime, or copy when you need ownership:
+// ScanRecord is a value type — safe to store and accumulate:
+std::vector<fluss::ScanRecord> all_records;
 fluss::ScanRecords records;
 scanner.Poll(5000, records);
 for (const auto& rec : records) {
+    all_records.push_back(rec);                    // safe! ref-counted
     auto name = rec.row.GetString(0);              // zero-copy string_view
     auto owned = std::string(rec.row.GetString(0)); // explicit copy when needed
-    process(owned);
 }
+
+// DON'T — string_view dangles after all records referencing the data are destroyed:
+std::string_view dangling;
+{
+    fluss::ScanRecords records;
+    scanner.Poll(5000, records);
+    dangling = records[0].row.GetString(0);
+}
+// dangling is undefined behavior here — no ScanRecord keeps the data alive!
 ```
 
 ### From Scan Results (RowView)

website/docs/user-guide/cpp/example/log-tables.md

@@ -62,6 +62,34 @@ for (const auto& rec : records) {
 }
 ```
 
+**Continuous polling:**
+
+```cpp
+while (running) {
+    fluss::ScanRecords records;
+    scanner.Poll(1000, records);
+    for (const auto& rec : records) {
+        process(rec);
+    }
+}
+```
+
+**Accumulating records across polls:**
+
+`ScanRecord` is a value type — it can be freely copied, stored, and accumulated. The underlying data stays alive via reference counting (zero-copy).
+
+```cpp
+std::vector<fluss::ScanRecord> all_records;
+while (all_records.size() < 1000) {
+    fluss::ScanRecords records;
+    scanner.Poll(1000, records);
+    for (const auto& rec : records) {
+        all_records.push_back(rec);  // ref-counted, no data copy
+    }
+}
+// all_records is valid — each record keeps its data alive
+```
+
 **Batch subscribe:**
 
 ```cpp