feat(rest): support query auth for REST catalog tables

include/paimon/defs.h

@@ -324,6 +324,12 @@ struct PAIMON_EXPORT Options {
     /// "rest.commit.lock-enabled" on the table. Default value is "false".
     static const char REST_COMMIT_ACQUIRE_LOCK[];
 
+    /// "query-auth.enabled" - Whether the catalog enforces row filters and column masks on reads
+    /// of the table. A scan of such a table asks the catalog built into its context for the rules
+    /// and attaches them to its splits, and a read applies them; a read path that would not apply
+    /// them is refused. The options of a read cannot disable it. Default value is "false".
+    static const char QUERY_AUTH_ENABLED[];
+
     /// "compaction.max-size-amplification-percent" - The size amplification is defined as the
     /// amount (in percentage) of additional storage needed to store a single byte of data in the
     /// merge tree for changelog mode table. Default value is 200.

include/paimon/read_context.h

@@ -408,6 +408,8 @@ class PAIMON_EXPORT ReadContextBuilder {
 
     /// Whether to perform precise filtering according to predicates for data read from format
     /// reader.
+    /// @note On a table with `query-auth.enabled=true` the predicate is evaluated on the masked
+    ///       values, and its conjuncts on masked columns are evaluated even when this is false.
     /// @param enabled Whether to enable precise filtering (default: false)
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& EnablePredicateFilter(bool enabled);
@@ -566,11 +568,16 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
     ReadContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
 
-    /// Reads a native table through its own file system - including the per-table temporary
-    /// credentials a catalog that issues them hands out through `Catalog::GetTableFileSystem`.
-    /// This is a shorthand for `WithFileSystem(catalog->GetTableFileSystem(identifier))`; an
-    /// explicit `WithFileSystem()` takes precedence, so the catalog is not asked.
-    /// @param catalog Non-null catalog, read when `Finish()` builds the context.
+    /// Reads a native table through its catalog, which supplies its file system:
+    /// `catalog->GetTableFileSystem(identifier)`, with any per-table temporary credentials; an
+    /// explicit `WithFileSystem()` takes precedence, so the catalog is not asked for one.
+    /// `TableRead::Create()` reads `query-auth.enabled` from the schema the catalog holds for the
+    /// branch read; the fields the read reads with still come from `SetTableSchema()` on the main
+    /// branch or from the latest schema under the table path. A branch the catalog would read as
+    /// another object, such as `MAIN` or one holding a `$`, is rejected before the catalog is
+    /// asked for anything.
+    /// @param catalog Non-null catalog, read when `Finish()` builds the context and when
+    ///        `TableRead::Create()` runs.
     /// @param identifier The native table to read.
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& WithCatalog(const std::shared_ptr<Catalog>& catalog,

include/paimon/scan_context.h

@@ -62,6 +62,22 @@ class PAIMON_EXPORT ScanContext {
                 const std::shared_ptr<Catalog>& catalog,
                 const std::optional<Identifier>& identifier);
 
+    /// Also takes the columns the query reads, see `ScanContextBuilder::SetReadFieldNames()`.
+    ScanContext(const std::string& path, bool is_streaming_mode, std::optional<int32_t> limit,
+                const std::shared_ptr<ScanFilter>& scan_filter,
+                const std::shared_ptr<GlobalIndexResult>& global_index_result,
+                const std::shared_ptr<RealtimeContext>& realtime_context,
+                const std::shared_ptr<MemoryPool>& memory_pool,
+                const std::shared_ptr<Executor>& executor,
+                const std::shared_ptr<FileSystem>& specific_file_system,
+                const std::optional<std::string>& table_schema,
+                const std::map<std::string, std::string>& options,
+                const std::shared_ptr<Cache>& cache,
+                const std::shared_ptr<FormatTable>& format_table,
+                const std::shared_ptr<Catalog>& catalog,
+                const std::optional<Identifier>& identifier,
+                const std::optional<std::vector<std::string>>& read_field_names);
+
     ~ScanContext();
 
     const std::string& GetPath() const {
@@ -127,6 +143,10 @@ class PAIMON_EXPORT ScanContext {
         return identifier_;
     }
 
+    const std::optional<std::vector<std::string>>& GetReadFieldNames() const {
+        return read_field_names_;
+    }
+
  private:
     std::string path_;
     bool is_streaming_mode_;
@@ -143,6 +163,7 @@ class PAIMON_EXPORT ScanContext {
     std::shared_ptr<FormatTable> format_table_;
     std::shared_ptr<Catalog> catalog_;
     std::optional<Identifier> identifier_;
+    std::optional<std::vector<std::string>> read_field_names_;
 };
 
 /// Filter configuration for table scan operations
@@ -197,6 +218,11 @@ class PAIMON_EXPORT ScanContextBuilder {
         const std::vector<std::map<std::string, std::string>>& partition_filters);
     /// Set a predicate for filtering data.
     ScanContextBuilder& SetPredicate(const std::shared_ptr<Predicate>& predicate);
+    /// Set the columns the query reads. A table with `query-auth.enabled=true` asks its catalog
+    /// for the rules of these columns plus those of the predicate and the partition filter; by
+    /// default it asks for all columns. An empty list asks for no other column, unlike the empty
+    /// list of `ReadContextBuilder::SetReadFieldNames()`, which reads every column.
+    ScanContextBuilder& SetReadFieldNames(const std::vector<std::string>& read_field_names);
 
     /// Sets the result of a global index search (e.g., row ids (may with scores) from a distributed
     /// index lookup). This is used to push down index-filtered row ids into the scan for efficient
@@ -241,11 +267,20 @@ class PAIMON_EXPORT ScanContextBuilder {
     /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
     ScanContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
 
-    /// Plans a native table through its own file system - including the per-table temporary
-    /// credentials a catalog that issues them hands out through `Catalog::GetTableFileSystem`.
-    /// This is a shorthand for `WithFileSystem(catalog->GetTableFileSystem(identifier))`; an
-    /// explicit `WithFileSystem()` takes precedence, so the catalog is not asked.
-    /// @param catalog Non-null catalog, read when `Finish()` builds the context.
+    /// Plans a native table through its catalog, which supplies its file system and, for a table
+    /// with `query-auth.enabled`, its rules. The file system is
+    /// `catalog->GetTableFileSystem(identifier)`, with any per-table temporary credentials; an
+    /// explicit `WithFileSystem()` takes precedence, so the catalog is not asked for one.
+    /// `TableScan::Create()` reads `query-auth.enabled` from the schema the catalog holds for the
+    /// branch scanned; the fields the scan plans with still come from `SetTableSchema()` on the
+    /// main branch or from the latest schema under the table path. Every plan of a query-auth
+    /// table checks that its selected and rule columns keep their field ids in the catalog's
+    /// current schema as well. A branch the `branch` option names is carried into the identifier,
+    /// as `tbl$branch_dev`; an identifier naming another branch, or a branch the catalog would
+    /// read as another object, such as `MAIN` or one holding a `$`, is rejected before the catalog
+    /// is asked for anything.
+    /// @param catalog Non-null catalog, read when `Finish()` builds the context and when
+    ///        `TableScan::Create()` runs.
     /// @param identifier The native table to scan.
     /// @return Reference to this builder for method chaining.
     ScanContextBuilder& WithCatalog(const std::shared_ptr<Catalog>& catalog,

include/paimon/table/source/split.h

@@ -39,7 +39,8 @@ class MemoryPool;
 /// This split can be a `DataSplit` (for direct data file reads), an `IndexedSplit` (for reads
 /// leveraging global indexes), or a `FormatDataSplit` (for a format table's plain data files).
 /// Only the first two are serializable; a `FormatDataSplit` is in-memory only and `Serialize()`
-/// refuses it.
+/// refuses it. A split of a table with `query-auth.enabled=true` wraps one of the first two with
+/// the row filters and column masks it is read with.
 class PAIMON_EXPORT Split {
  public:
     virtual ~Split() = default;

src/paimon/CMakeLists.txt

@@ -269,6 +269,7 @@ set(PAIMON_CORE_SRCS
     core/catalog/catalog_utils.cpp
     core/catalog/file_system_catalog.cpp
     core/catalog/identifier.cpp
+    core/catalog/table_query_auth_result.cpp
     core/core_options.cpp
     core/deletionvectors/deletion_vector.cpp
     core/deletionvectors/deletion_file_writer.cpp
@@ -480,6 +481,8 @@ set(PAIMON_CORE_SRCS
     core/table/source/merge_tree_split_generator.cpp
     core/table/source/data_evolution_split_generator.cpp
     core/table/source/plan_impl.cpp
+    core/table/source/query_auth_table_read.cpp
+    core/table/source/query_auth_table_scan.cpp
     core/table/source/realtime_table_scan.cpp
     core/table/source/snapshot/snapshot_reader.cpp
     core/table/source/startup_mode.cpp
@@ -846,6 +849,7 @@ if(PAIMON_BUILD_TESTS)
                     core/catalog/file_system_catalog_test.cpp
                     core/catalog/catalog_test.cpp
                     core/catalog/identifier_test.cpp
+                    core/catalog/table_query_auth_result_test.cpp
                     core/compact/compact_deletion_file_test.cpp
                     core/compact/compact_future_manager_test.cpp
                     core/compact/noop_compact_manager_test.cpp
@@ -1039,6 +1043,8 @@ if(PAIMON_BUILD_TESTS)
                     core/table/sink/commit_message_impl_test.cpp
                     core/table/source/fallback_data_split_test.cpp
                     core/table/source/primary_key_sorted_index_scan_test.cpp
+                    core/table/source/query_auth_split_test.cpp
+                    core/table/source/query_auth_table_read_test.cpp
                     core/table/source/table_read_test.cpp
                     core/table/source/append_count_reader_test.cpp
                     core/table/source/pk_count_reader_test.cpp

src/paimon/common/defs.cpp

@@ -91,6 +91,7 @@ const char Options::COMMIT_MAX_RETRY_WAIT[] = "commit.max-retry-wait";
 const char Options::COMMIT_DISCARD_DUPLICATE_FILES[] = "commit.discard-duplicate-files";
 const char Options::REST_COMMIT_LOCK_ENABLED[] = "rest.commit.lock-enabled";
 const char Options::REST_COMMIT_ACQUIRE_LOCK[] = "rest.commit.acquire-lock";
+const char Options::QUERY_AUTH_ENABLED[] = "query-auth.enabled";
 const char Options::SEQUENCE_FIELD[] = "sequence.field";
 const char Options::SEQUENCE_FIELD_SORT_ORDER[] = "sequence.field.sort-order";
 const char Options::MERGE_ENGINE[] = "merge-engine";

src/paimon/common/predicate/like.cpp

@@ -44,11 +44,6 @@ inline size_t Utf8CodePointLength(unsigned char leading_byte) {
     return 1;  // invalid continuation byte, treat as single byte
 }
 
-inline bool IsJavaRegexLineTerminator(const std::string& code_point) {
-    return code_point == "\n" || code_point == "\r" || code_point == "\xC2\x85" ||
-           code_point == "\xE2\x80\xA8" || code_point == "\xE2\x80\xA9";
-}
-
 }  // namespace
 
 Result<bool> Like::MatchString(const std::string& field, const std::string& pattern) {
@@ -153,7 +148,8 @@ Result<bool> Like::MatchString(const std::string& field, const std::string& patt
             if (wild && pc == "%") {
                 dp[j] = dp[j - 1] || dp[j];
             } else if (wild && pc == "_") {
-                dp[j] = prev && !IsJavaRegexLineTerminator(field_char);
+                // As Java `Like`, which matches `_` with `(?s:.)`: any one code point.
+                dp[j] = prev;
             } else {
                 dp[j] = (pc == field_char) ? prev : false;
             }

src/paimon/common/predicate/predicate_test.cpp

@@ -1765,26 +1765,30 @@ TEST_F(PredicateTest, TestLikeUtf8MultibyteUnderscore) {
     ASSERT_FALSE(predicate->Test(arrow_schema, CreateStringRow({"helloworld"})).value());
 }
 
-TEST_F(PredicateTest, TestLikeJavaRegexLineTerminatorSemantics) {
+TEST_F(PredicateTest, TestLikeWildcardsMatchLineTerminators) {
     auto arrow_schema = arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())}));
-
-    // Java regex '.' does not match line terminators, so '_' should not match them either.
-    ASSERT_OK_AND_ASSIGN(auto predicate_base,
-                         PredicateBuilder::Like(
-                             /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
-                             Literal(FieldType::STRING, "_", 1)));
-    auto predicate = std::dynamic_pointer_cast<PredicateFilter>(predicate_base);
-    ASSERT_FALSE(predicate->Test(arrow_schema, CreateStringRow({"\n"})).value());
-    ASSERT_FALSE(predicate->Test(arrow_schema, CreateStringRow({"\r"})).value());
-
-    // Java LIKE '%' uses (?s:.*), so it should still match line terminators.
-    ASSERT_OK_AND_ASSIGN(predicate_base,
-                         PredicateBuilder::Like(
-                             /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
-                             Literal(FieldType::STRING, "%", 1)));
-    predicate = std::dynamic_pointer_cast<PredicateFilter>(predicate_base);
-    ASSERT_TRUE(predicate->Test(arrow_schema, CreateStringRow({"\n"})).value());
-    ASSERT_TRUE(predicate->Test(arrow_schema, CreateStringRow({"\r"})).value());
+    auto like = [&](const std::string& pattern, const std::string& value) -> Result<bool> {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Predicate> predicate,
+                               PredicateBuilder::Like(
+                                   /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
+                                   Literal(FieldType::STRING, pattern.data(), pattern.size())));
+        auto filter = std::dynamic_pointer_cast<PredicateFilter>(predicate);
+        if (filter == nullptr) {
+            return Status::Invalid("the LIKE predicate is not a filter");
+        }
+        return filter->Test(arrow_schema, CreateStringRow({value}));
+    };
+    for (const std::string line_terminator :
+         {"\n", "\r", "\xC2\x85", "\xE2\x80\xA8", "\xE2\x80\xA9"}) {
+        ASSERT_OK_AND_ASSIGN(bool matched, like("a_b", "a" + line_terminator + "b"));
+        ASSERT_TRUE(matched) << line_terminator;
+        ASSERT_OK_AND_ASSIGN(matched, like("%", line_terminator));
+        ASSERT_TRUE(matched) << line_terminator;
+    }
+    ASSERT_OK_AND_ASSIGN(bool matched, like("a_b", "a\r\nb"));
+    ASSERT_FALSE(matched);
+    ASSERT_OK_AND_ASSIGN(matched, like("a__b", "a\r\nb"));
+    ASSERT_TRUE(matched);
 }
 
 TEST_F(PredicateTest, TestCompound) {

src/paimon/common/reader/blob_fallback_batch_reader_test.cpp

@@ -324,6 +324,7 @@ TEST_F(BlobFallbackBatchReaderTest, TestRejectsBatchExceedingReadLimit) {
         // This mock ignores the context and returns two rows for a one-row read/skip request.
         auto oversized_reader =
             std::make_unique<MockFileBatchReader>(MakeBlobStruct({"b0", "b1"}), struct_type_, 1024);
+        oversized_reader->EnableRandomizeBatchSize(false);
         std::vector<std::vector<BlobFallbackBatchReader::Segment>> groups;
         groups.push_back(MakeGroup(
             {SegmentSpec::File(covered ? BlobRows{"u0", "u1"} : BlobRows{"PH", "PH"})}, 1024));

src/paimon/common/types/data_type_json_parser.cpp

@@ -757,8 +757,9 @@ Result<std::shared_ptr<arrow::Field>> DataTypeJsonParser::ParseAtomicTypeField(
     const std::string& name, const rapidjson::Value& type_json_value) {
     bool nullable = true;
     AtomicTypeAttributes attributes;
+    const std::string type_str(type_json_value.GetString(), type_json_value.GetStringLength());
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::DataType> type,
-                           ParseAtomicType(type_json_value.GetString(), &nullable, &attributes));
+                           ParseAtomicType(type_str, &nullable, &attributes));
     if (attributes.is_blob) {
         return BlobUtils::ToArrowField(name, nullable);
     } else if (attributes.is_variant) {
@@ -776,11 +777,15 @@ Result<std::shared_ptr<arrow::Field>> DataTypeJsonParser::ParseAtomicTypeField(
 Result<std::shared_ptr<arrow::Field>> DataTypeJsonParser::ParseComplexTypeField(
     const std::string& name, const rapidjson::Value& type_json_value,
     FieldIdAssigner* field_id_assigner) {
-    if (!type_json_value.HasMember("type")) {
-        return Status::Invalid("complex data type must have type");
+    if (!type_json_value.HasMember("type") || !type_json_value["type"].IsString()) {
+        return Status::Invalid("complex data type must have a string type");
     }
 
-    std::string type_str = type_json_value["type"].GetString();
+    const auto& type_value = type_json_value["type"];
+    const std::string type_str(type_value.GetString(), type_value.GetStringLength());
+    if (type_str.find('\0') != std::string::npos) {
+        return Status::Invalid("complex data type cannot contain a NUL byte");
+    }
     bool nullable = true;
     if (type_str.find("NOT NULL") != std::string::npos) {
         nullable = false;

src/paimon/common/types/data_type_json_parser_test.cpp

@@ -228,8 +228,21 @@ TEST(DataTypeJsonParserTest, ParseTypeRowTypeGeneratesMissingFieldIds) {
     }
 }
 
+TEST(DataTypeJsonParserTest, ParseTypeRejectsEmbeddedNul) {
+    for (const char* json :
+         {R"("INT\u0000suffix")", R"({"type": "ARRAY\u0000suffix", "element": "INT"})",
+          R"({"type": "ARRAY", "element": "STRING\u0000suffix"})"}) {
+        rapidjson::Document doc;
+        doc.Parse(json);
+        ASSERT_FALSE(doc.HasParseError());
+        ASSERT_NOK(DataTypeJsonParser::ParseType("field", doc));
+    }
+}
+
 TEST(DataTypeJsonParserTest, ParseTypeComplexTypeFailure) {
     const std::vector<std::pair<const char*, const char*>> test_cases = {
+        {R"({"type":123})", "complex data type must have a string type"},
+        {R"({"element":"INT"})", "complex data type must have a string type"},
         {R"({"type":"ROW"})", "row data type must have fields"},
         {R"({"type":"ROW","fields":{}})", "row data type must have fields"},
         {R"({"type":"ROW","fields":[null]})", "data field must be an object"},