implement streaming read/write type info

Author: chaokunyang

cpp/fory/serialization/context.cc

@@ -64,46 +64,33 @@ WriteContext::WriteContext(const Config &config,
 
 WriteContext::~WriteContext() = default;
 
-Result<size_t, Error> WriteContext::push_meta(const std::type_index &type_id) {
-  auto it = write_type_id_index_map_.find(type_id);
-  if (it != write_type_id_index_map_.end()) {
-    return it->second;
-  }
-
-  size_t index = write_type_defs_.size();
+Result<void, Error>
+WriteContext::write_type_meta(const std::type_index &type_id) {
+  // Resolve type_index to TypeInfo* and delegate to the TypeInfo* version
+  // This ensures consistent indexing when the same type is written via
+  // either type_index or TypeInfo* path
   FORY_TRY(type_info, type_resolver_->get_type_info(type_id));
-  write_type_defs_.push_back(type_info->type_def);
-  write_type_id_index_map_[type_id] = index;
-  return index;
+  write_type_meta(type_info);
+  return Result<void, Error>();
 }
 
-size_t WriteContext::push_meta(const TypeInfo *type_info) {
+void WriteContext::write_type_meta(const TypeInfo *type_info) {
   auto it = write_type_info_index_map_.find(type_info);
   if (it != write_type_info_index_map_.end()) {
-    return it->second;
+    // Reference to previously written type: (index << 1) | 1, LSB=1
+    buffer_.WriteVarUint32(static_cast<uint32_t>((it->second << 1) | 1));
+    return;
   }
 
-  size_t index = write_type_defs_.size();
-  write_type_defs_.push_back(type_info->type_def);
+  // New type: index << 1, LSB=0, followed by TypeDef bytes inline
+  size_t index = write_type_info_index_map_.size();
+  buffer_.WriteVarUint32(static_cast<uint32_t>(index << 1));
   write_type_info_index_map_[type_info] = index;
-  return index;
-}
 
-void WriteContext::write_meta(size_t offset) {
-  size_t current_pos = buffer_.writer_index();
-  // Update the meta offset field (written as -1 initially)
-  int32_t meta_size = static_cast<int32_t>(current_pos - offset - 4);
-  buffer_.UnsafePut<int32_t>(offset, meta_size);
-  // Write all collected TypeMetas
-  buffer_.WriteVarUint32(static_cast<uint32_t>(write_type_defs_.size()));
-  for (size_t i = 0; i < write_type_defs_.size(); ++i) {
-    const auto &type_def = write_type_defs_[i];
-    buffer_.WriteBytes(type_def.data(), type_def.size());
-  }
+  // Write TypeDef bytes inline
+  buffer_.WriteBytes(type_info->type_def.data(), type_info->type_def.size());
 }
 
-bool WriteContext::meta_empty() const { return write_type_defs_.empty(); }
-
 /// Write pre-encoded meta string to buffer (avoids re-encoding on each write)
 static void write_encoded_meta_string(Buffer &buffer,
                                       const CachedMetaString &encoded) {
@@ -134,9 +121,8 @@ WriteContext::write_enum_typeinfo(const std::type_index &type) {
 
   if (type_id_low == static_cast<uint32_t>(TypeId::NAMED_ENUM)) {
     if (config_->compatible) {
-      // Write meta_index
-      FORY_TRY(meta_index, push_meta(type));
-      buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+      // Write type meta inline using streaming protocol
+      FORY_RETURN_NOT_OK(write_type_meta(type));
     } else {
       // Write pre-encoded namespace and type_name
       if (type_info->encoded_namespace && type_info->encoded_type_name) {
@@ -166,9 +152,8 @@ WriteContext::write_enum_typeinfo(const TypeInfo *type_info) {
 
   if (type_id_low == static_cast<uint32_t>(TypeId::NAMED_ENUM)) {
     if (config_->compatible) {
-      // Write meta_index using TypeInfo pointer (fast path)
-      size_t meta_index = push_meta(type_info);
-      buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+      // Write type meta inline using streaming protocol
+      write_type_meta(type_info);
     } else {
       // Write pre-encoded namespace and type_name
       if (type_info->encoded_namespace && type_info->encoded_type_name) {
@@ -207,18 +192,16 @@ WriteContext::write_any_typeinfo(uint32_t fory_type_id,
   switch (type_id_low) {
   case static_cast<uint32_t>(TypeId::NAMED_COMPATIBLE_STRUCT):
   case static_cast<uint32_t>(TypeId::COMPATIBLE_STRUCT): {
-    // Write meta_index
-    FORY_TRY(meta_index, push_meta(concrete_type_id));
-    buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+    // Write type meta inline using streaming protocol
+    FORY_RETURN_NOT_OK(write_type_meta(concrete_type_id));
     break;
   }
   case static_cast<uint32_t>(TypeId::NAMED_ENUM):
   case static_cast<uint32_t>(TypeId::NAMED_EXT):
   case static_cast<uint32_t>(TypeId::NAMED_STRUCT): {
     if (config_->compatible) {
-      // Write meta_index (share_meta is effectively compatible in C++)
-      FORY_TRY(meta_index, push_meta(concrete_type_id));
-      buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+      // Write type meta inline using streaming protocol
+      FORY_RETURN_NOT_OK(write_type_meta(concrete_type_id));
     } else {
       // Write pre-encoded namespace and type_name
       if (type_info->encoded_namespace && type_info->encoded_type_name) {
@@ -253,16 +236,14 @@ WriteContext::write_struct_type_info(const std::type_index &type_id) {
   switch (type_id_low) {
   case static_cast<uint32_t>(TypeId::NAMED_COMPATIBLE_STRUCT):
   case static_cast<uint32_t>(TypeId::COMPATIBLE_STRUCT): {
-    // Write meta_index
-    FORY_TRY(meta_index, push_meta(type_id));
-    buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+    // Write type meta inline using streaming protocol
+    FORY_RETURN_NOT_OK(write_type_meta(type_id));
     break;
   }
   case static_cast<uint32_t>(TypeId::NAMED_STRUCT): {
     if (config_->compatible) {
-      // Write meta_index
-      FORY_TRY(meta_index, push_meta(type_id));
-      buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+      // Write type meta inline using streaming protocol
+      FORY_RETURN_NOT_OK(write_type_meta(type_id));
     } else {
       // Write pre-encoded namespace and type_name
       if (type_info->encoded_namespace && type_info->encoded_type_name) {
@@ -295,16 +276,14 @@ WriteContext::write_struct_type_info(const TypeInfo *type_info) {
   switch (type_id_low) {
   case static_cast<uint32_t>(TypeId::NAMED_COMPATIBLE_STRUCT):
   case static_cast<uint32_t>(TypeId::COMPATIBLE_STRUCT): {
-    // Write meta_index using TypeInfo pointer (fast path)
-    size_t meta_index = push_meta(type_info);
-    buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+    // Write type meta inline using streaming protocol
+    write_type_meta(type_info);
     break;
   }
   case static_cast<uint32_t>(TypeId::NAMED_STRUCT): {
     if (config_->compatible) {
-      // Write meta_index using TypeInfo pointer (fast path)
-      size_t meta_index = push_meta(type_info);
-      buffer_.WriteVarUint32(static_cast<uint32_t>(meta_index));
+      // Write type meta inline using streaming protocol
+      write_type_meta(type_info);
     } else {
       // Write pre-encoded namespace and type_name
       if (type_info->encoded_namespace && type_info->encoded_type_name) {
@@ -329,10 +308,7 @@ void WriteContext::reset() {
   // Clear error state first
   error_ = Error();
   ref_writer_.reset();
-  // Clear meta vectors/maps - they're typically small or empty
-  // in non-compatible mode, so clear() is efficient
-  write_type_defs_.clear();
-  write_type_id_index_map_.clear();
+  // Clear meta map for streaming TypeMeta (size is used as counter)
   write_type_info_index_map_.clear();
   current_dyn_depth_ = 0;
   // Reset buffer indices for reuse - no memory operations needed
@@ -385,100 +361,94 @@ ReadContext::read_enum_type_info(uint32_t base_type_id) {
 // Maximum number of parsed type defs to cache (avoid OOM from malicious input)
 static constexpr size_t kMaxParsedNumTypeDefs = 8192;
 
-Result<size_t, Error> ReadContext::load_type_meta(int32_t meta_offset) {
-  size_t current_pos = buffer_->reader_index();
-  size_t meta_start = current_pos + meta_offset;
-  buffer_->ReaderIndex(static_cast<uint32_t>(meta_start));
-
-  // Load all TypeMetas
+Result<const TypeInfo *, Error> ReadContext::read_type_meta() {
   Error error;
-  uint32_t meta_size = buffer_->ReadVarUint32(error);
+  // Read the index marker
+  uint32_t index_marker = buffer_->ReadVarUint32(error);
   if (FORY_PREDICT_FALSE(!error.ok())) {
     return Unexpected(std::move(error));
   }
-  reading_type_infos_.reserve(meta_size);
 
-  for (uint32_t i = 0; i < meta_size; i++) {
-    // Read the 8-byte header first for caching
-    int64_t meta_header = buffer_->ReadInt64(error);
-    if (FORY_PREDICT_FALSE(!error.ok())) {
-      return Unexpected(std::move(error));
-    }
+  bool is_ref = (index_marker & 1) == 1;
+  size_t index = index_marker >> 1;
 
-    // Check if we already parsed this type meta (cache lookup by header)
-    auto cache_it = parsed_type_infos_.find(meta_header);
-    if (cache_it != parsed_type_infos_.end()) {
-      // Found in cache - reuse and skip the bytes
-      reading_type_infos_.push_back(cache_it->second);
-      FORY_RETURN_NOT_OK(TypeMeta::skip_bytes(*buffer_, meta_header));
-      continue;
-    }
+  if (is_ref) {
+    // Reference to previously read type
+    return get_type_info_by_index(index);
+  }
 
-    // Not in cache - parse the TypeMeta
-    FORY_TRY(parsed_meta,
-             TypeMeta::from_bytes_with_header(*buffer_, meta_header));
-
-    // Find local TypeInfo to get field_id mapping (optional for schema
-    // evolution)
-    const TypeInfo *local_type_info = nullptr;
-    if (parsed_meta->register_by_name) {
-      auto result = type_resolver_->get_type_info_by_name(
-          parsed_meta->namespace_str, parsed_meta->type_name);
-      if (result.ok()) {
-        local_type_info = result.value();
-      }
-    } else {
-      auto result = type_resolver_->get_type_info_by_id(parsed_meta->type_id);
-      if (result.ok()) {
-        local_type_info = result.value();
-      }
-    }
+  // New type - read TypeMeta inline
+  // Read the 8-byte header first for caching
+  int64_t meta_header = buffer_->ReadInt64(error);
+  if (FORY_PREDICT_FALSE(!error.ok())) {
+    return Unexpected(std::move(error));
+  }
 
-    // Create TypeInfo with field_ids assigned
-    auto type_info = std::make_unique<TypeInfo>();
-    if (local_type_info) {
-      // Have local type - assign field_ids by comparing schemas
-      // Note: Extension types don't have type_meta (only structs do)
-      if (local_type_info->type_meta) {
-        TypeMeta::assign_field_ids(local_type_info->type_meta.get(),
-                                   parsed_meta->field_infos);
-      }
-      type_info->type_id = local_type_info->type_id;
-      type_info->type_meta = std::move(parsed_meta);
-      type_info->type_def = local_type_info->type_def;
-      // CRITICAL: Copy the harness from the registered type_info
-      type_info->harness = local_type_info->harness;
-      type_info->name_to_index = local_type_info->name_to_index;
-      type_info->namespace_name = local_type_info->namespace_name;
-      type_info->type_name = local_type_info->type_name;
-      type_info->register_by_name = local_type_info->register_by_name;
-    } else {
-      // No local type - create stub TypeInfo with parsed meta
-      type_info->type_id = parsed_meta->type_id;
-      type_info->type_meta = std::move(parsed_meta);
+  // Check if we already parsed this type meta (cache lookup by header)
+  auto cache_it = parsed_type_infos_.find(meta_header);
+  if (cache_it != parsed_type_infos_.end()) {
+    // Found in cache - reuse and skip the bytes
+    reading_type_infos_.push_back(cache_it->second);
+    FORY_RETURN_NOT_OK(TypeMeta::skip_bytes(*buffer_, meta_header));
+    return cache_it->second;
+  }
+
+  // Not in cache - parse the TypeMeta
+  FORY_TRY(parsed_meta,
+           TypeMeta::from_bytes_with_header(*buffer_, meta_header));
+
+  // Find local TypeInfo to get field_id mapping (optional for schema evolution)
+  const TypeInfo *local_type_info = nullptr;
+  if (parsed_meta->register_by_name) {
+    auto result = type_resolver_->get_type_info_by_name(
+        parsed_meta->namespace_str, parsed_meta->type_name);
+    if (result.ok()) {
+      local_type_info = result.value();
+    }
+  } else {
+    auto result = type_resolver_->get_type_info_by_id(parsed_meta->type_id);
+    if (result.ok()) {
+      local_type_info = result.value();
     }
+  }
 
-    // Get raw pointer before moving into storage
-    const TypeInfo *raw_ptr = type_info.get();
+  // Create TypeInfo with field_ids assigned
+  auto type_info = std::make_unique<TypeInfo>();
+  if (local_type_info) {
+    // Have local type - assign field_ids by comparing schemas
+    // Note: Extension types don't have type_meta (only structs do)
+    if (local_type_info->type_meta) {
+      TypeMeta::assign_field_ids(local_type_info->type_meta.get(),
+                                 parsed_meta->field_infos);
+    }
+    type_info->type_id = local_type_info->type_id;
+    type_info->type_meta = std::move(parsed_meta);
+    type_info->type_def = local_type_info->type_def;
+    // CRITICAL: Copy the harness from the registered type_info
+    type_info->harness = local_type_info->harness;
+    type_info->name_to_index = local_type_info->name_to_index;
+    type_info->namespace_name = local_type_info->namespace_name;
+    type_info->type_name = local_type_info->type_name;
+    type_info->register_by_name = local_type_info->register_by_name;
+  } else {
+    // No local type - create stub TypeInfo with parsed meta
+    type_info->type_id = parsed_meta->type_id;
+    type_info->type_meta = std::move(parsed_meta);
+  }
 
-    // Store in primary storage
-    owned_reading_type_infos_.push_back(std::move(type_info));
+  // Get raw pointer before moving into storage
+  const TypeInfo *raw_ptr = type_info.get();
 
-    // Cache the parsed TypeInfo (with size limit to prevent OOM)
-    if (parsed_type_infos_.size() < kMaxParsedNumTypeDefs) {
-      parsed_type_infos_[meta_header] = raw_ptr;
-    }
+  // Store in primary storage
+  owned_reading_type_infos_.push_back(std::move(type_info));
 
-    reading_type_infos_.push_back(raw_ptr);
+  // Cache the parsed TypeInfo (with size limit to prevent OOM)
+  if (parsed_type_infos_.size() < kMaxParsedNumTypeDefs) {
+    parsed_type_infos_[meta_header] = raw_ptr;
   }
 
-  // Calculate size of meta section
-  size_t meta_end = buffer_->reader_index();
-  size_t meta_section_size = meta_end - meta_start;
-
-  // Restore buffer position
-  buffer_->ReaderIndex(static_cast<uint32_t>(current_pos));
-  return meta_section_size;
+  reading_type_infos_.push_back(raw_ptr);
+  return raw_ptr;
 }
 
 Result<const TypeInfo *, Error>
@@ -499,25 +469,19 @@ Result<const TypeInfo *, Error> ReadContext::read_any_typeinfo() {
   }
   uint32_t type_id_low = type_id & 0xff;
 
-  // Mirror Rust's read_any_typeinfo using switch for jump table generation
+  // Use streaming protocol for type meta
   switch (type_id_low) {
   case static_cast<uint32_t>(TypeId::NAMED_COMPATIBLE_STRUCT):
   case static_cast<uint32_t>(TypeId::COMPATIBLE_STRUCT): {
-    uint32_t meta_index = buffer_->ReadVarUint32(error);
-    if (FORY_PREDICT_FALSE(!error.ok())) {
-      return Unexpected(std::move(error));
-    }
-    return get_type_info_by_index(meta_index);
+    // Read type meta inline using streaming protocol
+    return read_type_meta();
   }
   case static_cast<uint32_t>(TypeId::NAMED_ENUM):
   case static_cast<uint32_t>(TypeId::NAMED_EXT):
   case static_cast<uint32_t>(TypeId::NAMED_STRUCT): {
     if (config_->compatible) {
-      uint32_t meta_index = buffer_->ReadVarUint32(error);
-      if (FORY_PREDICT_FALSE(!error.ok())) {
-        return Unexpected(std::move(error));
-      }
-      return get_type_info_by_index(meta_index);
+      // Read type meta inline using streaming protocol
+      return read_type_meta();
     }
     FORY_TRY(namespace_str,
              meta_string_table_.read_string(*buffer_, kNamespaceDecoder));