perf(embedding): initialize logical input and result payloads

docs/mo-exact-decimal-integration.md

@@ -6,6 +6,30 @@ The semantic authority remains MO #29449 document blob
 `42a89f09a1d168d02b9583cb3ea7b4de6dbb5634`; MO #29690 records the B-F
 delivery sequence. The detailed implementation plan was approved before code.
 
+## Approved input initialization optimization (2026-10-13)
+
+The MO owner approved a focused performance increment before implementation.
+Native input acquisition initializes exactly the admitted logical payload,
+after allocating and validating its physical storage and before returning the
+lease. Unwritten payload bytes remain zero. Rounded allocator padding stays
+unaddressable by the input descriptor and is never uploaded; all physical blocks
+and descriptors remain charged to the existing input window.
+
+Initialization belongs to the common acquisition owner, including CPU fixtures,
+rather than the GPU pool constructor. No input credit, coalescing, ABI, worker or
+allocator change is needed. A poisoned multi-block fixture proves cross-block
+zeroing, untouched padding and partial-failure credit release; real GPU input,
+cancellation/reuse and memory checks close the production consumer. The negation
+is exposure of stale logical bytes, initialization of unused padding, or retained
+credit after an allocation/initialization failure.
+
+The pinned pool also backs results and publisher scratch. Result acquisition
+uses the same logical-range initializer so alignment gaps and unwritten result
+bytes preserve the previous zero guarantee. CPU/poisoned result fixtures and
+real result packing close this reverse consumer; allocator padding remains
+charged and unaddressable. The initializer is a nonvirtual internal storage
+helper and changes no C ABI or storage ownership.
+
 ## Contracts and owners
 
 The validated import descriptor owns width, precision, scale and nullability.

src/embedding/input.cpp

@@ -26,6 +26,13 @@ T load(input_storage const& data, std::size_t offset)
   return value;
 }
 }  // namespace
+void input_storage::initialize(std::size_t bytes)
+{
+  range(0, bytes, size());
+  visit([bytes](std::size_t offset, std::span<std::byte> block) {
+    if (offset < bytes) std::memset(block.data(), 0, std::min(block.size(), bytes - offset));
+  });
+}
 void input_storage::write(std::size_t offset, std::span<const std::byte> bytes)
 {
   range(offset, bytes.size(), size());
@@ -230,6 +237,10 @@ std::shared_ptr<input_batch> native_input::acquire(std::size_t bytes, clock::tim
   batch->storage       = pool->allocate(allocated);
   batch->payload_bytes = bytes;
   require(batch->storage && batch->storage->size() == allocated, "invalid native pool allocation");
+  // Initialize every addressable byte, including fields a caller leaves
+  // unwritten. Physical pool padding stays charged but is never exposed or
+  // uploaded, so clearing it would amplify work for selective reader batches.
+  batch->storage->initialize(bytes);
   {
     std::lock_guard lock(mutex_);
     check_open();

src/embedding/input.hpp

@@ -20,6 +20,7 @@ struct input_storage {
   virtual std::size_t size() const                                                  = 0;
   virtual void visit(std::function<void(std::size_t, std::span<std::byte>)> const&) = 0;
   void write(std::size_t offset, std::span<const std::byte> bytes);
+  void initialize(std::size_t bytes);
   virtual void read(std::size_t offset, std::span<std::byte> bytes) const;
 };
 struct input_pool {

src/embedding/native_gpu.cpp

@@ -13,12 +13,7 @@ namespace {
 using host_resource = cucascade::memory::fixed_size_host_memory_resource;
 struct pool_storage final : input_storage {
   host_resource::fixed_multiple_blocks_allocation blocks;
-  explicit pool_storage(host_resource::fixed_multiple_blocks_allocation p) : blocks(std::move(p))
-  {
-    // Never expose bytes from a previous query when a caller leaves a field unwritten.
-    for (auto* block : blocks->get_blocks())
-      std::memset(block, 0, blocks->block_size());
-  }
+  explicit pool_storage(host_resource::fixed_multiple_blocks_allocation p) : blocks(std::move(p)) {}
   std::size_t size() const override { return blocks->size_bytes(); }
   void read(std::size_t offset, std::span<std::byte> bytes) const override
   {

src/embedding/result.cpp

@@ -67,7 +67,10 @@ sirius_status native_result::try_allocate(std::size_t bytes,
   if (batch->stats) batch->stats->result_retain(charge);
   batch->credit = std::move(credit);
   batch->pool   = pool;
-  if (bytes) batch->storage = pool->allocate(bytes);
+  if (bytes) {
+    batch->storage = pool->allocate(bytes);
+    batch->storage->initialize(bytes);
+  }
   batch->columns       = std::make_unique<sirius_input_vector[]>(columns);
   batch->column_count  = columns;
   batch->payload_bytes = bytes;

test/cpp/embedding/test_native_input.cpp

@@ -74,6 +74,34 @@ struct rounded_pool : heap_pool {
     return (bytes + block - 1) / block * block;
   }
 };
+struct poisoned_storage : input_storage {
+  std::shared_ptr<std::vector<std::byte>> data;
+  std::size_t block;
+  bool fail;
+  poisoned_storage(std::shared_ptr<std::vector<std::byte>> bytes, std::size_t size, bool failure)
+    : data(std::move(bytes)), block(size), fail(failure)
+  {
+  }
+  std::size_t size() const override { return data->size(); }
+  void visit(std::function<void(std::size_t, std::span<std::byte>)> const& fn) override
+  {
+    for (std::size_t offset = 0; offset < data->size(); offset += block) {
+      if (fail && offset) throw std::bad_alloc();
+      fn(offset, std::span{*data}.subspan(offset, block));
+    }
+  }
+};
+struct poisoned_pool : rounded_pool {
+  std::weak_ptr<std::vector<std::byte>> last;
+  bool fail{false};
+  poisoned_pool() : rounded_pool(32) {}
+  std::unique_ptr<input_storage> allocate(std::size_t bytes) override
+  {
+    auto data = std::make_shared<std::vector<std::byte>>(bytes, std::byte{0xa5});
+    last      = data;
+    return std::make_unique<poisoned_storage>(std::move(data), block, fail);
+  }
+};
 struct observed_storage : heap_storage {
   std::shared_ptr<std::atomic<int>> reads;
   std::shared_ptr<std::atomic<bool>> fail;
@@ -111,6 +139,35 @@ void publish_ints(input_fixture& f, uint32_t rows, int64_t first = 0)
   f.input->publish(batch, rows, {&column, 1});
 }
 }  // namespace
+TEST_CASE("native input initializes logical bytes without clearing pool padding", "[native_input]")
+{
+  auto pool = std::make_shared<poisoned_pool>();
+  input_fixture f({23, 0, 0, 1}, 64 + column_charge, input_target, 4, pool);
+  auto batch = f.input->acquire(43, clock::now());
+  {
+    auto bytes = pool->last.lock();
+    REQUIRE(bytes);
+    REQUIRE(bytes->size() == 64);
+    for (std::size_t i = 0; i < bytes->size(); ++i)
+      CHECK((*bytes)[i] == (i < 43 ? std::byte{0} : std::byte{0xa5}));
+  }
+  CHECK(f.input->inspect().retained_bytes == 64 + column_charge);
+  std::byte value{1};
+  REQUIRE_THROWS_AS(batch->write(43, {&value, 1}), failure);
+  sirius_input_vector column{};
+  column.data_offset = 40;
+  column.data_bytes  = 8;
+  REQUIRE_THROWS_AS(f.input->publish(batch, 1, {&column, 1}), failure);
+  batch.reset();
+  CHECK(pool->last.expired());
+  CHECK(f.input->inspect().retained_bytes == 0);
+
+  pool->fail = true;
+  REQUIRE_THROWS_AS(f.input->acquire(43, clock::now()), std::bad_alloc);
+  CHECK(pool->last.expired());
+  CHECK(f.input->inspect().retained_bytes == 0);
+  CHECK(f.input->inspect().filling_batches == 0);
+}
 TEST_CASE("native input credits follow filling queued and retry owners", "[native_input]")
 {
   input_fixture f({23, 0, 0, 1}, 144, 140);

test/cpp/embedding/test_native_result.cpp

@@ -11,7 +11,7 @@ using namespace std::chrono_literals;
 namespace {
 struct result_test_storage final : input_storage {
   std::vector<std::byte> data;
-  explicit result_test_storage(std::size_t n) : data(n) {}
+  explicit result_test_storage(std::size_t n) : data(n, std::byte{0xa5}) {}
   std::size_t size() const override { return data.size(); }
   void visit(std::function<void(std::size_t, std::span<std::byte>)> const& fn) override
   {
@@ -33,6 +33,21 @@ struct result_fixture {
   ~result_fixture() { result->cancel(); }
 };
 }  // namespace
+TEST_CASE("native results initialize logical gaps without clearing allocation padding",
+          "[native_result]")
+{
+  result_fixture f;
+  std::shared_ptr<result_batch> batch;
+  REQUIRE(f.result->try_allocate(43, 0, batch) == SIRIUS_OK);
+  REQUIRE(batch->payload_bytes == 43);
+  std::array<std::byte, 48> bytes{};
+  batch->storage->read(0, bytes);
+  for (std::size_t i = 0; i < bytes.size(); ++i)
+    CHECK(bytes[i] == (i < batch->payload_bytes ? std::byte{0} : std::byte{0xa5}));
+  batch.reset();
+  CHECK(f.result->inspect().retained_bytes == 0);
+  CHECK(f.result->inspect().filling_batches == 0);
+}
 TEST_CASE("native result credit spans filling queued borrowed and cancellation", "[native_result]")
 {
   result_fixture f;