feat(realtime): add in-memory predicate indexes

docs/source/user_guide/realtime.rst

@@ -115,6 +115,11 @@ Common Options
      - ``none``
      - ``none`` or ``full``. Full mode collects per-batch min/max and null counts
        for conservative predicate pruning, including retained spill metadata.
+   * - ``realtime.index.<index_name>.columns``
+     - Unset
+     - Comma-separated fields indexed in memory by the default Arrow store.
+       Available in append and append deduplicate modes; independent of
+       ``realtime.store.stats-mode``.
    * - ``realtime.deduplicate-key-fields``
      - Unset
      - Selects append deduplicate mode. Although expressed as a comma-separated
@@ -398,6 +403,37 @@ context's reclaimed progress for relevant active stores. It does not restore
 an old in-memory state and is not unrestricted historical time travel. For a
 disk-only historical query, do not attach the real-time context.
 
+In-memory Predicate Indexes
+--------------------------
+
+The default Arrow store supports optional predicate indexes on key or value fields
+in append and append deduplicate modes. PK modes are unsupported. For example:
+
+.. code-block:: text
+
+   realtime.enabled = true
+   realtime.index.hash.columns = id,payload
+   realtime.store.stats-mode = full
+
+The built-in ``hash`` plugin is **experimental**; its implementation and supported
+capabilities may change. It supports equality and ``IN`` predicates on boolean,
+integer, floating-point, string, binary, and date fields. Decimal, timestamp, and
+nested types are currently unsupported. Use a comma-separated list to index
+multiple fields; indexing ``_REALTIME_OFFSET`` is unsupported.
+
+Note that this index does not currently handle hash collisions.
+
+Indexes and batch statistics work independently and can be enabled together.
+For ``id = 7 AND score > 100``, an index supporting equality on ``id`` selects
+candidate offsets; statistics can then skip batches whose maximum ``score``
+is at most 100.
+Statistics cover entire batches, so mixed batches still need row-level
+filtering. Keep predicate fields in the read schema and enable
+``ReadContextBuilder::EnablePredicateFilter(true)`` for exact filtering.
+
+In-memory and disk indexes are configured independently. Use
+``file-index.<type>.columns`` to enable indexes for committed files.
+
 Implementation by Mode
 ----------------------
 
@@ -406,9 +442,10 @@ Append
 
 ``RealtimeAppendOnlyWriter`` validates offsets and writes to an
 ``APPEND_ONLY`` store. The default ``ArrowRealtimeStore`` retains building
-batches and immutable sealed segments in write order. Query readers project
-the requested fields, optionally prune stored batches using statistics, and
-apply the visible offset range before removing the hidden offset column.
+batches and immutable sealed segments in write order. Query readers use
+optional predicate indexes and batch statistics to prune candidates, project
+the requested fields, and apply the visible offset range before removing the
+hidden offset column.
 
 Preparing reads sealed segments through the ordinary ``AppendOnlyWriter``,
 removing the internal offset prefix before writing table files. Queries do not
@@ -467,8 +504,10 @@ are marked deleted, and within-batch duplicates are resolved in row order.
 Add rows are appended to the store; delete events update only the key/DV state,
 without persisting a delete row.
 
-Key indexes and DV bitmaps use copy-on-write snapshots, so later writes do not
-change the state held by existing lookups, read views, or seal checkpoints.
+Deduplicate key-to-latest-offset lookups and DV bitmaps use copy-on-write
+snapshots, so later writes do not change the state held by existing lookups,
+read views, or seal checkpoints. The optional predicate indexes share
+append-only postings instead.
 
 Memory reads use the read view's DV to exclude rows whose offsets are marked
 deleted.
@@ -511,8 +550,9 @@ still terminal for the owning writer/context.
 Old views can retain the memory representation after spill or reclamation;
 new views use the published representation. Spill files are deleted when the
 store and all remaining handles/readers have released them. They are temporary
-storage, not a recoverable log after process restart. Deduplicate key indexes
-and DVs remain in memory; full-mode statistics also remain resident after spill.
+storage, not a recoverable log after process restart. Predicate indexes,
+deduplicate key indexes, DVs, and full-mode statistics remain in memory after
+spill. Predicate index memory is included in store memory accounting.
 
 ``RealtimeContext::GetMetrics()`` and real-time writer metrics expose:
 
@@ -596,8 +636,8 @@ Current restrictions include:
 * Append deduplicate requires a single non-null scalar key, a persisted offset
   column, exact offset file-index reads, and non-bitmap64 DVs. Existing ordinary
   append files are not migrated automatically.
-* Spill covers Arrow data, not key indexes or DVs, and provides neither a hard
-  memory limit nor crash recovery.
+* Spill covers Arrow data, not predicate indexes, key indexes, or DVs, and
+  provides neither a hard memory limit nor crash recovery.
 
 Code Map
 --------

include/paimon/api.h

@@ -42,6 +42,9 @@
 #include "paimon/write_context.h"                // IWYU pragma: export
 
 // IWYU pragma: begin_exports
+#include "paimon/realtime/index/realtime_index.h"
+#include "paimon/realtime/index/realtime_index_factory.h"
+#include "paimon/realtime/index/realtime_index_reader.h"
 #include "paimon/realtime/realtime_context.h"
 #include "paimon/realtime/realtime_snapshot_offsets.h"
 #include "paimon/realtime/realtime_store.h"

include/paimon/file_store_write.h

@@ -41,6 +41,9 @@ class RecordBatch;
 class WriteContext;
 
 /// Interface for write operations in a file store.
+///
+/// A `RealtimeContext` supports only one active writer for each partition-bucket. Close the
+/// previous writer before creating a replacement that accesses the same partition-bucket.
 class PAIMON_EXPORT FileStoreWrite {
  public:
     /// Create an instance of `FileStoreWrite`.
@@ -134,6 +137,10 @@ class PAIMON_EXPORT FileStoreWrite {
     ///
     /// The writer loads the snapshot's partition-bucket offsets and releases sealed memory that is
     /// fully covered by disk. Calling this method on a non-real-time writer returns an error.
+    /// After a successful refresh, fully covered segments are also removed from this writer's
+    /// pending prepare queue. Existing read views continue to pin the resources they reference.
+    /// Snapshot installation and queue cleanup are serialized with `Seal()` and
+    /// `PrepareCommitWithProgress()`.
     ///
     /// With `WriteContextBuilder::WithCatalog()`, the latest snapshot can be loaded from the
     /// catalog. Older requested snapshots must be available on the file system.
@@ -162,8 +169,9 @@ class PAIMON_EXPORT FileStoreWrite {
 
     /// Releases resources owned by this writer.
     ///
-    /// Closing a real-time writer with data not covered by a successful
-    /// `PrepareCommitWithProgress()` returns an error and invalidates its `RealtimeContext`.
+    /// Closing a real-time writer with data covered by neither a successful
+    /// `PrepareCommitWithProgress()` nor a refreshed committed snapshot returns an error and
+    /// invalidates its `RealtimeContext`.
     /// The caller must rebuild both objects and recover input from the durable snapshot offset.
     virtual Status Close() = 0;
 };

include/paimon/realtime/arrow_realtime_store_factory.h

@@ -28,6 +28,8 @@ class PAIMON_EXPORT ArrowRealtimeStoreFactory : public RealtimeStoreFactory {
  public:
     /// Creates an Arrow-backed append or primary-key store for one partition and bucket.
     /// The framework uses the append implementation as the delegate of its DEDUPLICATE wrapper.
+    /// Append delegates resolve configured index plugins through the registry.
+    /// Primary-key stores do not support query indexes.
     Result<std::shared_ptr<RealtimeStore>> Create(RealtimeStoreCreateRequest&& request) override;
 };
 

include/paimon/realtime/index/realtime_index.h

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+
+#include "paimon/realtime/offset_range.h"
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+struct ArrowArray;
+
+namespace paimon {
+class RealtimeIndexReader;
+
+/// Append-only in-memory index for one field, supporting one writer and concurrent readers.
+class PAIMON_EXPORT RealtimeIndex {
+ public:
+    virtual ~RealtimeIndex() = default;
+
+    /// Consumes [_REALTIME_OFFSET, field] with increasing offsets.
+    virtual Status AddBatch(::ArrowArray* batch) = 0;
+
+    /// Creates a reader for [begin, end), sharing postings without copying.
+    virtual Result<std::shared_ptr<RealtimeIndexReader>> CreateReader(
+        const OffsetRange& offset_range) const = 0;
+
+    virtual uint64_t GetMemoryUsage() const = 0;
+};
+
+}  // namespace paimon

include/paimon/realtime/index/realtime_index_factory.h

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <map>
+#include <memory>
+#include <string>
+
+#include "paimon/factories/factory.h"
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+struct ArrowSchema;
+
+namespace paimon {
+class MemoryPool;
+class RealtimeIndex;
+
+/// Factory for field indexes configured by realtime.index.<identifier>.columns.
+class PAIMON_EXPORT RealtimeIndexFactory : public Factory {
+ public:
+    ~RealtimeIndexFactory() override = default;
+
+    /// Registry suffix for real-time index plugins.
+    static const char REALTIME_INDEX_IDENTIFIER_SUFFIX[];
+
+    /// Resolves identifier + "-realtime" and consumes the [_REALTIME_OFFSET, field] schema.
+    static Result<std::shared_ptr<RealtimeIndex>> Get(
+        const std::string& identifier, ::ArrowSchema* write_schema,
+        const std::map<std::string, std::string>& options, const std::shared_ptr<MemoryPool>& pool);
+
+    /// Creates a new field index; consumes the [_REALTIME_OFFSET, field] schema.
+    virtual Result<std::shared_ptr<RealtimeIndex>> Create(
+        ::ArrowSchema* write_schema, const std::map<std::string, std::string>& options,
+        const std::shared_ptr<MemoryPool>& pool) const = 0;
+};
+
+}  // namespace paimon

include/paimon/realtime/index/realtime_index_reader.h

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <vector>
+
+#include "paimon/predicate/function_visitor.h"
+#include "paimon/utils/roaring_bitmap64.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// Visits predicates on one indexed field, returning realtime offsets.
+/// nullopt means no pruning; an empty bitmap means no candidates. False negatives are not allowed.
+/// Unsupported predicates default to no pruning. Plugins override only the supported visits.
+class PAIMON_EXPORT RealtimeIndexReader : public FunctionVisitor<std::optional<RoaringBitmap64>> {
+ public:
+    ~RealtimeIndexReader() override = default;
+
+    Result<std::optional<RoaringBitmap64>> VisitIsNotNull() override;
+    Result<std::optional<RoaringBitmap64>> VisitIsNull() override;
+    Result<std::optional<RoaringBitmap64>> VisitEqual(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitNotEqual(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitLessThan(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitLessOrEqual(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitGreaterThan(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitGreaterOrEqual(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitIn(const std::vector<Literal>& literals) override;
+    Result<std::optional<RoaringBitmap64>> VisitNotIn(
+        const std::vector<Literal>& literals) override;
+    Result<std::optional<RoaringBitmap64>> VisitStartsWith(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitEndsWith(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitContains(const Literal& literal) override;
+    Result<std::optional<RoaringBitmap64>> VisitLike(const Literal& literal) override;
+};
+
+}  // namespace paimon

include/paimon/realtime/realtime_context.h

@@ -88,6 +88,8 @@ class PAIMON_EXPORT RealtimeMetrics {
 /// reads. `RealtimeContext` itself is not a customization interface and must not be implemented by
 /// applications. Customize real-time storage and retrieval through `RealtimeStoreFactory` and
 /// `RealtimeStore` instead.
+/// Only one writer may actively use a partition-bucket in this context. Close that writer before
+/// creating its replacement; writers for different partition-buckets may share the context.
 ///
 /// A context is valid only for one uninterrupted committed-progress history. Overwrite, truncate,
 /// partition drop, and rollback operations do not automatically clear process-local real-time

include/paimon/realtime/realtime_store.h

@@ -142,6 +142,9 @@ struct PAIMON_EXPORT RealtimeQueryContext {
     /// Optional predicate using field indexes from `read_schema`. A non-null predicate allows the
     /// plugin to prune candidate rows. Exact filtering is applied by the Paimon read framework.
     std::shared_ptr<Predicate> predicate;
+    /// Required [disk snapshot committed end, pinned memory end) bounds.
+    /// Pin memory before selecting the disk snapshot; keep these bounds for the query.
+    OffsetRange offset_range;
     /// Maximum number of rows in a returned query batch.
     int32_t read_batch_size = 1024;
 };

src/paimon/CMakeLists.txt

@@ -424,6 +424,11 @@ set(PAIMON_CORE_SRCS
     core/realtime/arrow_realtime_store_factory.cpp
     core/realtime/data_file_key_offset_index_writer.cpp
     core/realtime/key_offset_index_utils.cpp
+    core/realtime/index/hash/hash_realtime_index.cpp
+    core/realtime/index/realtime_index_evaluator.cpp
+    core/realtime/index/realtime_index_factory.cpp
+    core/realtime/index/realtime_index_reader.cpp
+    core/realtime/index/realtime_index_set.cpp
     core/realtime/realtime_key_offset_index.cpp
     core/realtime/realtime_file_schema.cpp
     core/realtime/realtime_offset_file_index_lookup.cpp
@@ -914,6 +919,7 @@ if(PAIMON_BUILD_TESTS)
                     core/realtime/arrow_deduplicate_realtime_store_test.cpp
                     core/realtime/arrow_realtime_store_test.cpp
                     core/realtime/key_offset_index_utils_test.cpp
+                    core/realtime/index/realtime_index_test.cpp
                     core/realtime/primary_key_realtime_store_test.cpp
                     core/realtime/realtime_deduplicate_state_test.cpp
                     core/realtime/realtime_file_schema_test.cpp

src/paimon/common/io/offset_output_stream.cpp

@@ -39,6 +39,14 @@ Result<int64_t> OffsetOutputStream::GetPos() const {
     return position - start_;
 }
 
+Result<int64_t> OffsetOutputStream::GetUnderlyingPos() const {
+    const OutputStream* output = output_;
+    while (const auto* offset_output = dynamic_cast<const OffsetOutputStream*>(output)) {
+        output = offset_output->output_;
+    }
+    return output->GetPos();
+}
+
 Result<std::string> OffsetOutputStream::GetUri() const {
     return output_->GetUri();
 }

src/paimon/common/io/offset_output_stream.h

@@ -39,6 +39,8 @@ class PAIMON_EXPORT OffsetOutputStream : public OutputStream {
 
     Result<int64_t> Write(const char* buffer, int64_t size) override;
     Result<int64_t> GetPos() const override;
+    /// Returns the current position of the innermost stream, unwrapping nested OffsetOutputStreams.
+    Result<int64_t> GetUnderlyingPos() const;
     Result<std::string> GetUri() const override;
     Status Flush() override;
     Status Close() override;