fix comments

Author: gripleaf

docs/source/user_guide/system_tables.rst

@@ -38,6 +38,12 @@ from multiple LSM levels during query planning and reading. The tradeoff is that
 the result may lag behind the latest committed data until a full compaction
 publishes the newest records into the highest level.
 
+This stale view is not guaranteed to correspond to any single historical table
+snapshot. Full compaction may finish independently for different buckets. When
+``$ro`` scans the latest snapshot, the selected highest-level files can
+therefore combine bucket states produced by full-compaction commits at different
+snapshot IDs.
+
 For primary-key tables, ``$ro`` also enables value-stats filtering, so file-level
 pruning of the reader predicate can be more aggressive than the base table.
 
@@ -48,7 +54,8 @@ Limitations
 ~~~~~~~~~~~
 
 - Primary-key ``$ro`` scans are batch-only. Streaming scans are not supported.
-- Freshness depends on full compaction frequency.
+- Freshness depends on full compaction frequency, and the result may not match
+  any single historical snapshot across buckets.
 - Primary-key tables in bucket-unaware mode are not supported by the current C++
   scan path.
 

src/paimon/core/table/system/read_optimized_system_table.cpp

@@ -20,6 +20,7 @@
 #include <string>
 #include <utility>
 
+#include "arrow/c/bridge.h"
 #include "paimon/common/types/data_field.h"
 #include "paimon/core/schema/table_schema.h"
 #include "paimon/core/table/source/read_optimized_scan_options.h"
@@ -110,6 +111,13 @@ Result<std::unique_ptr<TableRead>> ReadOptimizedSystemTable::NewRead(
         .WithCache(context->GetCache())
         .SetReadFieldNames(context->GetReadFieldNames())
         .SetReadFieldIds(context->GetReadFieldIds());
+    if (context->HasReadSchema()) {
+        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> read_schema,
+                                          arrow::ImportSchema(context->GetReadSchema()));
+        auto c_read_schema = std::make_unique<::ArrowSchema>();
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema, c_read_schema.get()));
+        builder.SetReadSchema(std::move(c_read_schema));
+    }
     if (context->GetSpecificTableSchema().has_value()) {
         builder.SetTableSchema(context->GetSpecificTableSchema().value());
     }

test/inte/read_inte_test.cpp

@@ -985,6 +985,70 @@ TEST(SystemTableReadInteTest, TestReadOptimizedPrimaryKeyProjectionAndPredicateP
     ASSERT_TRUE(result->Equals(expected)) << result->ToString();
 }
 
+TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableNestedProjection) {
+    auto payload_type =
+        arrow::struct_({arrow::field("a", arrow::int32()), arrow::field("b", arrow::utf8())});
+    arrow::FieldVector fields = {
+        arrow::field("k", arrow::int32()),
+        arrow::field("payload", payload_type),
+        arrow::field("extra", arrow::int32()),
+    };
+    auto schema = arrow::schema(fields);
+    std::map<std::string, std::string> options = {{Options::FILE_SYSTEM, "local"},
+                                                  {Options::FILE_FORMAT, "parquet"},
+                                                  {Options::MANIFEST_FORMAT, "avro"},
+                                                  {Options::BUCKET, "1"},
+                                                  {Options::BUCKET_KEY, "k"}};
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TestHelper> helper,
+                         TestHelper::Create(dir->Str(), schema,
+                                            /*partition_keys=*/{}, /*primary_keys=*/{"k"}, options,
+                                            /*is_streaming_mode=*/true));
+    std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar");
+
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<RecordBatch> batch,
+        TestHelper::MakeRecordBatch(arrow::struct_(fields),
+                                    R"([[1, [10, "x"], 100], [2, [20, "y"], 200]])",
+                                    /*partition_map=*/{}, /*bucket=*/0, {}));
+    ASSERT_OK(WriteAndFullCompact(std::move(batch), /*commit_identifier=*/0, helper.get()));
+
+    ScanContextBuilder scan_context_builder(table_path + "$ro");
+    scan_context_builder.SetOptions(options);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, scan_context_builder.Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan,
+                         TableScan::Create(std::move(scan_context)));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan());
+
+    auto projected_schema = arrow::schema({
+        arrow::field("k", arrow::int32()),
+        arrow::field("payload", arrow::struct_({arrow::field("a", arrow::int32())})),
+    });
+    auto c_projected_schema = std::make_unique<ArrowSchema>();
+    ASSERT_TRUE(arrow::ExportSchema(*projected_schema, c_projected_schema.get()).ok());
+    ReadContextBuilder read_context_builder(table_path + "$ro");
+    read_context_builder.SetOptions(options).SetReadSchema(std::move(c_projected_schema));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read,
+                         TableRead::Create(std::move(read_context)));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> batch_reader,
+                         table_read->CreateReader(plan->Splits()));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> result,
+                         ReadResultCollector::CollectResult(batch_reader.get()));
+
+    std::shared_ptr<arrow::DataType> expected_type = arrow::struct_({
+        arrow::field("_VALUE_KIND", arrow::int8()),
+        arrow::field("k", arrow::int32()),
+        arrow::field("payload", arrow::struct_({arrow::field("a", arrow::int32())})),
+    });
+    std::shared_ptr<arrow::ChunkedArray> expected;
+    ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(
+                    expected_type, {R"([[0, 1, [10]], [0, 2, [20]]])"}, &expected)
+                    .ok());
+    ASSERT_TRUE(result->Equals(expected)) << result->ToString();
+}
+
 TEST(SystemTableReadInteTest, TestReadOptimizedSystemTableWithBranch) {
     arrow::FieldVector fields = {
         arrow::field("k", arrow::int32()),