Merge branch 'master' into async-loader-workloads

docs/en/operations/system-tables/row_policies.md

@@ -12,7 +12,7 @@ Columns:
 
 - `database` ([String](../../sql-reference/data-types/string.md)) — Database name.
 
-- `table` ([String](../../sql-reference/data-types/string.md)) — Table name.
+- `table` ([String](../../sql-reference/data-types/string.md)) — Table name. Empty if policy for database.
 
 - `id` ([UUID](../../sql-reference/data-types/uuid.md)) — Row policy ID.
 

docs/en/sql-reference/functions/date-time-functions.md

@@ -1218,12 +1218,16 @@ Rounds the time to the half hour.
 
 Converts a date or date with time to a UInt32 number containing the year and month number (YYYY \* 100 + MM). Accepts a second optional timezone argument. If provided, the timezone must be a string constant.
 
-### example
-```sql
+**Example**
+
+``` sql
 SELECT
     toYYYYMM(now(), 'US/Eastern')
 ```
-```response
+
+Result:
+
+``` text
 ┌─toYYYYMM(now(), 'US/Eastern')─┐
 │                        202303 │
 └───────────────────────────────┘
@@ -1233,11 +1237,15 @@ SELECT
 
 Converts a date or date with time to a UInt32 number containing the year and month number (YYYY \* 10000 + MM \* 100 + DD). Accepts a second optional timezone argument. If provided, the timezone must be a string constant.
 
-### example
+**Example**
+
 ```sql
 SELECT
     toYYYYMMDD(now(), 'US/Eastern')
 ```
+
+Result:
+
 ```response
 ┌─toYYYYMMDD(now(), 'US/Eastern')─┐
 │                        20230302 │
@@ -1248,11 +1256,15 @@ SELECT
 
 Converts a date or date with time to a UInt64 number containing the year and month number (YYYY \* 10000000000 + MM \* 100000000 + DD \* 1000000 + hh \* 10000 + mm \* 100 + ss). Accepts a second optional timezone argument. If provided, the timezone must be a string constant.
 
-### example
+**Example**
+
 ```sql
 SELECT
     toYYYYMMDDhhmmss(now(), 'US/Eastern')
 ```
+
+Result:
+
 ```response
 ┌─toYYYYMMDDhhmmss(now(), 'US/Eastern')─┐
 │                        20230302112209 │

docs/en/sql-reference/statements/create/row-policy.md

@@ -14,8 +14,8 @@ Row policies makes sense only for users with readonly access. If user can modify
 Syntax:
 
 ``` sql
-CREATE [ROW] POLICY [IF NOT EXISTS | OR REPLACE] policy_name1 [ON CLUSTER cluster_name1] ON [db1.]table1
-        [, policy_name2 [ON CLUSTER cluster_name2] ON [db2.]table2 ...]
+CREATE [ROW] POLICY [IF NOT EXISTS | OR REPLACE] policy_name1 [ON CLUSTER cluster_name1] ON [db1.]table1|db1.*
+        [, policy_name2 [ON CLUSTER cluster_name2] ON [db2.]table2|db2.* ...]
     [FOR SELECT] USING condition
     [AS {PERMISSIVE | RESTRICTIVE}]
     [TO {role1 [, role2 ...] | ALL | ALL EXCEPT role1 [, role2 ...]}]
@@ -76,6 +76,20 @@ CREATE ROW POLICY pol2 ON mydb.table1 USING c=2 AS RESTRICTIVE TO peter, antonio
 
 enables the user `peter` to see rows only if both `b=1` AND `c=2`.
 
+Database policies are combined with table policies.
+
+For example, the following policies
+
+``` sql
+CREATE ROW POLICY pol1 ON mydb.* USING b=1 TO mira, peter
+CREATE ROW POLICY pol2 ON mydb.table1 USING c=2 AS RESTRICTIVE TO peter, antonio
+```
+
+enables the user `peter` to see table1 rows only if both `b=1` AND `c=2`, although
+any other table in mydb would have only `b=1` policy applied for the user.
+
+
+
 ## ON CLUSTER Clause
 
 Allows creating row policies on a cluster, see [Distributed DDL](../../../sql-reference/distributed-ddl.md).
@@ -88,3 +102,5 @@ Allows creating row policies on a cluster, see [Distributed DDL](../../../sql-re
 `CREATE ROW POLICY filter2 ON mydb.mytable USING a<1000 AND b=5 TO ALL EXCEPT mira`
 
 `CREATE ROW POLICY filter3 ON mydb.mytable USING 1 TO admin`
+
+`CREATE ROW POLICY filter4 ON mydb.* USING 1 TO admin`

src/Access/AuthenticationData.cpp

@@ -10,6 +10,7 @@
 #include <Common/OpenSSLHelpers.h>
 #include <Poco/SHA1Engine.h>
 #include <base/types.h>
+#include <base/hex.h>
 #include <boost/algorithm/hex.hpp>
 #include <boost/algorithm/string/case_conv.hpp>
 

src/Access/Common/RowPolicyDefs.cpp

@@ -22,7 +22,7 @@ String RowPolicyName::toString() const
         name += backQuoteIfNeed(database);
         name += '.';
     }
-    name += backQuoteIfNeed(table_name);
+    name += (table_name == RowPolicyName::ANY_TABLE_MARK ? "*" : backQuoteIfNeed(table_name));
     return name;
 }
 

src/Access/Common/RowPolicyDefs.h

@@ -9,6 +9,8 @@ namespace DB
 /// Represents the full name of a row policy, e.g. "myfilter ON mydb.mytable".
 struct RowPolicyName
 {
+    static constexpr char ANY_TABLE_MARK[] = "";
+
     String short_name;
     String database;
     String table_name;

src/Access/EnabledRowPolicies.cpp

@@ -35,7 +35,13 @@ RowPolicyFilterPtr EnabledRowPolicies::getFilter(const String & database, const
     auto loaded = mixed_filters.load();
     auto it = loaded->find({database, table_name, filter_type});
     if (it == loaded->end())
-        return {};
+    {   /// Look for a policy for database if a table policy not found
+        it = loaded->find({database, RowPolicyName::ANY_TABLE_MARK, filter_type});
+        if (it == loaded->end())
+        {
+            return {};
+        }
+    }
 
     return it->second;
 }

src/Access/RolesOrUsersSet.cpp

@@ -228,25 +228,25 @@ void RolesOrUsersSet::add(const std::vector<UUID> & ids_)
 
 bool RolesOrUsersSet::match(const UUID & id) const
 {
-    return (all || ids.count(id)) && !except_ids.count(id);
+    return (all || ids.contains(id)) && !except_ids.contains(id);
 }
 
 
 bool RolesOrUsersSet::match(const UUID & user_id, const boost::container::flat_set<UUID> & enabled_roles) const
 {
-    if (!all && !ids.count(user_id))
+    if (!all && !ids.contains(user_id))
     {
         bool found_enabled_role = std::any_of(
-            enabled_roles.begin(), enabled_roles.end(), [this](const UUID & enabled_role) { return ids.count(enabled_role); });
+            enabled_roles.begin(), enabled_roles.end(), [this](const UUID & enabled_role) { return ids.contains(enabled_role); });
         if (!found_enabled_role)
             return false;
     }
 
-    if (except_ids.count(user_id))
+    if (except_ids.contains(user_id))
         return false;
 
     bool in_except_list = std::any_of(
-        enabled_roles.begin(), enabled_roles.end(), [this](const UUID & enabled_role) { return except_ids.count(enabled_role); });
+        enabled_roles.begin(), enabled_roles.end(), [this](const UUID & enabled_role) { return except_ids.contains(enabled_role); });
     return !in_except_list;
 }
 

src/Access/RowPolicy.h

@@ -35,6 +35,9 @@ struct RowPolicy : public IAccessEntity
     void setPermissive(bool permissive_ = true) { setRestrictive(!permissive_); }
     bool isPermissive() const { return !isRestrictive(); }
 
+    /// Applied for entire database
+    bool isForDatabase() const { return full_name.table_name == RowPolicyName::ANY_TABLE_MARK; }
+
     /// Sets that the policy is restrictive.
     /// A row is only accessible if at least one of the permissive policies passes,
     /// in addition to all the restrictive policies.

src/Access/RowPolicyCache.cpp

@@ -16,7 +16,8 @@ namespace DB
 {
 namespace
 {
-    /// Accumulates filters from multiple row policies and joins them using the AND logical operation.
+    /// Helper to accumulate filters from multiple row policies and join them together
+    ///   by AND or OR logical operations.
     class FiltersMixer
     {
     public:
@@ -148,9 +149,11 @@ void RowPolicyCache::ensureAllRowPoliciesRead()
 
     for (const UUID & id : access_control.findAll<RowPolicy>())
     {
-        auto quota = access_control.tryRead<RowPolicy>(id);
-        if (quota)
-            all_policies.emplace(id, PolicyInfo(quota));
+        auto policy = access_control.tryRead<RowPolicy>(id);
+        if (policy)
+        {
+            all_policies.emplace(id, PolicyInfo(policy));
+        }
     }
 }
 
@@ -215,40 +218,105 @@ void RowPolicyCache::mixFiltersFor(EnabledRowPolicies & enabled)
         std::vector<RowPolicyPtr> policies;
     };
 
-    std::unordered_map<MixedFiltersKey, MixerWithNames, Hash> mixers;
+    std::unordered_map<MixedFiltersKey, MixerWithNames, Hash> database_mixers;
 
+    /// populate database_mixers using database-level policies
+    ///  to aggregate (mix) rules per database
     for (const auto & [policy_id, info] : all_policies)
     {
-        const auto & policy = *info.policy;
-        bool match = info.roles->match(enabled.params.user_id, enabled.params.enabled_roles);
-        MixedFiltersKey key;
-        key.database = info.database_and_table_name->first;
-        key.table_name = info.database_and_table_name->second;
-        for (auto filter_type : collections::range(0, RowPolicyFilterType::MAX))
+        if (info.isForDatabase())
         {
-            auto filter_type_i = static_cast<size_t>(filter_type);
-            if (info.parsed_filters[filter_type_i])
+            const auto & policy = *info.policy;
+            bool match = info.roles->match(enabled.params.user_id, enabled.params.enabled_roles);
+            for (auto filter_type : collections::range(0, RowPolicyFilterType::MAX))
             {
-                key.filter_type = filter_type;
-                auto & mixer = mixers[key];
-                mixer.database_and_table_name = info.database_and_table_name;
-                if (match)
+                auto filter_type_i = static_cast<size_t>(filter_type);
+                if (info.parsed_filters[filter_type_i])
                 {
-                    mixer.mixer.add(info.parsed_filters[filter_type_i], policy.isRestrictive());
-                    mixer.policies.push_back(info.policy);
+                    MixedFiltersKey key{info.database_and_table_name->first,
+                        info.database_and_table_name->second,
+                        filter_type};
+
+                    auto & mixer = database_mixers[key];
+                    mixer.database_and_table_name = info.database_and_table_name;
+                    if (match)
+                    {
+                        mixer.mixer.add(info.parsed_filters[filter_type_i], policy.isRestrictive());
+                        mixer.policies.push_back(info.policy);
+                    }
+                }
+            }
+        }
+    }
+
+    std::unordered_map<MixedFiltersKey, MixerWithNames, Hash> table_mixers;
+
+    /// populate table_mixers using database_mixers and table-level policies
+    for (const auto & [policy_id, info] : all_policies)
+    {
+        if (!info.isForDatabase())
+        {
+            const auto & policy = *info.policy;
+            bool match = info.roles->match(enabled.params.user_id, enabled.params.enabled_roles);
+            for (auto filter_type : collections::range(0, RowPolicyFilterType::MAX))
+            {
+                auto filter_type_i = static_cast<size_t>(filter_type);
+                if (info.parsed_filters[filter_type_i])
+                {
+                    MixedFiltersKey key{info.database_and_table_name->first,
+                        info.database_and_table_name->second,
+                        filter_type};
+                    auto table_it = table_mixers.find(key);
+                    if (table_it == table_mixers.end())
+                    {   /// no exact match - create new mixer
+                        MixedFiltersKey database_key = key;
+                        database_key.table_name = RowPolicyName::ANY_TABLE_MARK;
+
+                        auto database_it = database_mixers.find(database_key);
+
+                        if (database_it == database_mixers.end())
+                        {
+                            table_it = table_mixers.try_emplace(key).first;
+                        }
+                        else
+                        {
+                            /// table policies are based on database ones
+                            table_it = table_mixers.insert({key, database_it->second}).first;
+                        }
+                    }
+
+                    auto & mixer = table_it->second; ///  getting table level mixer
+                    mixer.database_and_table_name = info.database_and_table_name;
+                    if (match)
+                    {
+                        mixer.mixer.add(info.parsed_filters[filter_type_i], policy.isRestrictive());
+                        mixer.policies.push_back(info.policy);
+                    }
                 }
             }
         }
     }
 
     auto mixed_filters = boost::make_shared<MixedFiltersMap>();
-    for (auto & [key, mixer] : mixers)
+
+    /// Retrieve aggregated policies from mixers
+    ///  if a table has a policy for this particular table, we have all needed information in table_mixers
+    ///    (policies for the database are already applied)
+    ///  otherwise we would look for a policy for database using RowPolicy::ANY_TABLE_MARK
+    /// Consider restrictive policies a=1 for db.t, b=2 for db.* and c=3 for db.*
+    ///   We are going to have two items in mixed_filters:
+    ///     1. a=1 AND b=2 AND c=3   for db.t (comes from table_mixers, where it had been created with the help of database_mixers)
+    ///     2. b=2 AND c=3  for db.* (comes directly from database_mixers)
+    for (auto * mixer_map_ptr : {&table_mixers, &database_mixers})
     {
-        auto mixed_filter = std::make_shared<RowPolicyFilter>();
-        mixed_filter->database_and_table_name = std::move(mixer.database_and_table_name);
-        mixed_filter->expression = std::move(mixer.mixer).getResult(access_control.isEnabledUsersWithoutRowPoliciesCanReadRows());
-        mixed_filter->policies = std::move(mixer.policies);
-        mixed_filters->emplace(key, std::move(mixed_filter));
+        for (auto & [key, mixer] : *mixer_map_ptr)
+        {
+            auto mixed_filter = std::make_shared<RowPolicyFilter>();
+            mixed_filter->database_and_table_name = std::move(mixer.database_and_table_name);
+            mixed_filter->expression = std::move(mixer.mixer).getResult(access_control.isEnabledUsersWithoutRowPoliciesCanReadRows());
+            mixed_filter->policies = std::move(mixer.policies);
+            mixed_filters->emplace(key, std::move(mixed_filter));
+        }
     }
 
     enabled.mixed_filters.store(mixed_filters);

src/Access/RowPolicyCache.h

@@ -29,6 +29,7 @@ private:
         explicit PolicyInfo(const RowPolicyPtr & policy_) { setPolicy(policy_); }
         void setPolicy(const RowPolicyPtr & policy_);
 
+        bool isForDatabase() const { return policy->isForDatabase(); }
         RowPolicyPtr policy;
         const RolesOrUsersSet * roles = nullptr;
         std::shared_ptr<const std::pair<String, String>> database_and_table_name;

src/AggregateFunctions/AggregateFunctionForEach.h

@@ -2,6 +2,7 @@
 
 #include <Columns/ColumnArray.h>
 #include <Common/assert_cast.h>
+#include <Common/Arena.h>
 #include <base/arithmeticOverflow.h>
 #include <DataTypes/DataTypeArray.h>
 #include <AggregateFunctions/IAggregateFunction.h>

src/AggregateFunctions/AggregateFunctionMinMaxAny.h

@@ -10,6 +10,7 @@
 #include <DataTypes/IDataType.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <base/StringRef.h>
+#include <Common/Arena.h>
 #include <Common/assert_cast.h>
 #include <DataTypes/DataTypeNullable.h>
 #include <AggregateFunctions/IAggregateFunction.h>

src/CMakeLists.txt

@@ -211,6 +211,7 @@ endif()
 if (TARGET ch_contrib::jemalloc)
     target_link_libraries (clickhouse_common_io PRIVATE ch_contrib::jemalloc)
 endif()
+target_link_libraries (clickhouse_common_io PUBLIC ch_contrib::sparsehash)
 
 add_subdirectory(Access/Common)
 add_subdirectory(Common/ZooKeeper)
@@ -463,7 +464,7 @@ endif ()
 if (TARGET ch_contrib::ldap)
     dbms_target_link_libraries (PRIVATE ch_contrib::ldap ch_contrib::lber)
 endif ()
-dbms_target_link_libraries (PRIVATE ch_contrib::sparsehash)
+dbms_target_link_libraries (PUBLIC ch_contrib::sparsehash)
 
 if (TARGET ch_contrib::protobuf)
     dbms_target_link_libraries (PRIVATE ch_contrib::protobuf)

src/Core/Block.cpp

@@ -663,12 +663,10 @@ Names Block::getDataTypeNames() const
 
 Block::NameMap Block::getNamesToIndexesMap() const
 {
-    NameMap res;
-    res.reserve(index_by_name.size());
-
+    NameMap res(index_by_name.size());
+    res.set_empty_key(StringRef{});
     for (const auto & [name, index] : index_by_name)
         res[name] = index;
-
     return res;
 }
 

src/Core/Block.h

@@ -5,13 +5,11 @@
 #include <Core/ColumnsWithTypeAndName.h>
 #include <Core/NamesAndTypes.h>
 
-#include <Common/HashTable/HashMap.h>
-
 #include <initializer_list>
 #include <list>
-#include <map>
 #include <set>
 #include <vector>
+#include <sparsehash/dense_hash_map>
 
 
 namespace DB
@@ -97,7 +95,7 @@ public:
     Names getDataTypeNames() const;
 
     /// Hash table match `column name -> position in the block`.
-    using NameMap = HashMap<StringRef, size_t, StringRefHash>;
+    using NameMap = ::google::dense_hash_map<StringRef, size_t, StringRefHash>;
     NameMap getNamesToIndexesMap() const;
 
     Serializations getSerializations() const;

src/Core/Settings.h

@@ -748,6 +748,7 @@ class IColumn;
     M(Bool, optimize_distinct_in_order, true, "Enable DISTINCT optimization if some columns in DISTINCT form a prefix of sorting. For example, prefix of sorting key in merge tree or ORDER BY statement", 0) \
     M(Bool, allow_experimental_undrop_table_query, false, "Allow to use undrop query to restore dropped table in a limited time", 0) \
     M(Bool, keeper_map_strict_mode, false, "Enforce additional checks during operations on KeeperMap. E.g. throw an exception on an insert for already existing key", 0) \
+    M(UInt64, extract_kvp_max_pairs_per_row, 1000, "Max number pairs that can be produced by extractKeyValuePairs function. Used to safeguard against consuming too much memory.", 0) \
     // End of COMMON_SETTINGS
     // Please add settings related to formats into the FORMAT_FACTORY_SETTINGS and move obsolete settings to OBSOLETE_SETTINGS.
 

src/DataTypes/Serializations/SerializationAggregateFunction.cpp

@@ -7,6 +7,7 @@
 #include <Common/typeid_cast.h>
 #include <Common/assert_cast.h>
 #include <Common/AlignedBuffer.h>
+#include <Common/Arena.h>
 
 #include <Formats/FormatSettings.h>
 #include <Formats/ProtobufReader.h>

src/Disks/IO/AsynchronousReadIndirectBufferFromRemoteFS.cpp

@@ -118,12 +118,7 @@ std::future<IAsynchronousReader::Result> AsynchronousReadIndirectBufferFromRemot
     request.size = size;
     request.offset = file_offset_of_buffer_end;
     request.priority = base_priority + priority;
-
-    if (bytes_to_ignore)
-    {
-        request.ignore = bytes_to_ignore;
-        bytes_to_ignore = 0;
-    }
+    request.ignore = bytes_to_ignore;
     return reader.submit(request);
 }
 
@@ -165,8 +160,7 @@ void AsynchronousReadIndirectBufferFromRemoteFS::setReadUntilPosition(size_t pos
 
 void AsynchronousReadIndirectBufferFromRemoteFS::setReadUntilEnd()
 {
-    read_until_position = impl->getFileSize();
-    impl->setReadUntilPosition(*read_until_position);
+    setReadUntilPosition(impl->getFileSize());
 }
 
 
@@ -228,12 +222,13 @@ bool AsynchronousReadIndirectBufferFromRemoteFS::nextImpl()
 
         chassert(memory.size() == read_settings.prefetch_buffer_size || memory.size() == read_settings.remote_fs_buffer_size);
         std::tie(size, offset) = impl->readInto(memory.data(), memory.size(), file_offset_of_buffer_end, bytes_to_ignore);
-        bytes_to_ignore = 0;
 
         ProfileEvents::increment(ProfileEvents::RemoteFSUnprefetchedReads);
         ProfileEvents::increment(ProfileEvents::RemoteFSUnprefetchedBytes, size);
     }
 
+    bytes_to_ignore = 0;
+
     chassert(size >= offset);
 
     size_t bytes_read = size - offset;
@@ -269,7 +264,7 @@ off_t AsynchronousReadIndirectBufferFromRemoteFS::seek(off_t offset, int whence)
     }
     else if (whence == SEEK_CUR)
     {
-        new_pos = file_offset_of_buffer_end - (working_buffer.end() - pos) + offset;
+        new_pos = static_cast<size_t>(getPosition()) + offset;
     }
     else
     {
@@ -277,13 +272,15 @@ off_t AsynchronousReadIndirectBufferFromRemoteFS::seek(off_t offset, int whence)
     }
 
     /// Position is unchanged.
-    if (new_pos + (working_buffer.end() - pos) == file_offset_of_buffer_end)
+    if (new_pos == static_cast<size_t>(getPosition()))
         return new_pos;
 
     bool read_from_prefetch = false;
     while (true)
     {
-        if (file_offset_of_buffer_end - working_buffer.size() <= new_pos && new_pos <= file_offset_of_buffer_end)
+        /// The first condition implies bytes_to_ignore = 0.
+        if (!working_buffer.empty() && file_offset_of_buffer_end - working_buffer.size() <= new_pos &&
+            new_pos <= file_offset_of_buffer_end)
         {
             /// Position is still inside the buffer.
             /// Probably it is at the end of the buffer - then we will load data on the following 'next' call.
@@ -320,6 +317,7 @@ off_t AsynchronousReadIndirectBufferFromRemoteFS::seek(off_t offset, int whence)
 
     /// First reset the buffer so the next read will fetch new data to the buffer.
     resetWorkingBuffer();
+    bytes_to_ignore = 0;
 
     if (read_until_position && new_pos > *read_until_position)
     {
@@ -356,6 +354,12 @@ off_t AsynchronousReadIndirectBufferFromRemoteFS::seek(off_t offset, int whence)
 }
 
 
+off_t AsynchronousReadIndirectBufferFromRemoteFS::getPosition()
+{
+    return file_offset_of_buffer_end - available() + bytes_to_ignore;
+}
+
+
 void AsynchronousReadIndirectBufferFromRemoteFS::finalize()
 {
     resetPrefetch(FilesystemPrefetchState::UNNEEDED);

src/Disks/IO/AsynchronousReadIndirectBufferFromRemoteFS.h

@@ -42,7 +42,7 @@ public:
 
     off_t seek(off_t offset_, int whence) override;
 
-    off_t getPosition() override { return file_offset_of_buffer_end - available(); }
+    off_t getPosition() override;
 
     String getFileName() const override;
 
@@ -89,6 +89,8 @@ private:
 
     std::string current_reader_id;
 
+    /// If nonzero then working_buffer is empty.
+    /// If a prefetch is in flight, the prefetch task has been instructed to ignore this many bytes.
     size_t bytes_to_ignore = 0;
 
     std::optional<size_t> read_until_position;

src/Formats/ColumnMapping.cpp

@@ -26,8 +26,8 @@ void ColumnMapping::addColumns(
     {
         names_of_columns.push_back(name);
 
-        const auto * column_it = column_indexes_by_names.find(name);
-        if (!column_it)
+        const auto column_it = column_indexes_by_names.find(name);
+        if (column_it == column_indexes_by_names.end())
         {
             if (settings.skip_unknown_fields)
             {
@@ -43,7 +43,7 @@ void ColumnMapping::addColumns(
                             name, column_indexes_for_input_fields.size());
         }
 
-        const auto column_index = column_it->getMapped();
+        const auto column_index = column_it->second;
 
         if (read_columns[column_index])
             throw Exception(ErrorCodes::INCORRECT_DATA, "Duplicate field found while parsing format header: {}", name);

src/Functions/FunctionBinaryArithmetic.h

@@ -41,6 +41,7 @@
 #include <Common/FieldVisitorsAccurateComparison.h>
 #include <Common/assert_cast.h>
 #include <Common/typeid_cast.h>
+#include <Common/Arena.h>
 #include <DataTypes/DataTypeLowCardinality.h>
 #include <Interpreters/Context.h>
 

src/Functions/bitCount.cpp

@@ -9,7 +9,7 @@ namespace DB
 template <typename A>
 struct BitCountImpl
 {
-    using ResultType = UInt8;
+    using ResultType = std::conditional_t<(sizeof(A) * 8 >= 256), UInt16, UInt8>;
     static constexpr bool allow_string_or_fixed_string = true;
 
     static inline ResultType apply(A a)
@@ -17,6 +17,13 @@ struct BitCountImpl
         /// We count bits in the value representation in memory. For example, we support floats.
         /// We need to avoid sign-extension when converting signed numbers to larger type. So, uint8_t(-1) has 8 bits.
 
+        if constexpr (is_big_int_v<A>)
+        {
+            ResultType res = 0;
+            for (auto item : a.items)
+                res += __builtin_popcountll(item);
+            return res;
+        }
         if constexpr (std::is_same_v<A, UInt64> || std::is_same_v<A, Int64>)
             return __builtin_popcountll(a);
         if constexpr (std::is_same_v<A, UInt32> || std::is_same_v<A, Int32> || std::is_unsigned_v<A>)

src/Functions/keyvaluepair/extractKeyValuePairs.cpp

@@ -7,6 +7,8 @@
 #include <DataTypes/DataTypeMap.h>
 #include <DataTypes/DataTypeString.h>
 
+#include <Interpreters/Context.h>
+
 #include <Functions/keyvaluepair/impl/KeyValuePairExtractor.h>
 #include <Functions/keyvaluepair/impl/KeyValuePairExtractorBuilder.h>
 #include <Functions/keyvaluepair/ArgumentExtractor.h>
@@ -41,6 +43,13 @@ class ExtractKeyValuePairs : public IFunction
             builder.withQuotingCharacter(parsed_arguments.quoting_character.value());
         }
 
+        bool is_number_of_pairs_unlimited = context->getSettingsRef().extract_kvp_max_pairs_per_row == 0;
+
+        if (!is_number_of_pairs_unlimited)
+        {
+            builder.withMaxNumberOfPairs(context->getSettingsRef().extract_kvp_max_pairs_per_row);
+        }
+
         return builder.build();
     }
 
@@ -73,7 +82,7 @@ class ExtractKeyValuePairs : public IFunction
     }
 
 public:
-    ExtractKeyValuePairs() = default;
+    explicit ExtractKeyValuePairs(ContextPtr context_) : context(context_) {}
 
     static constexpr auto name = Name::name;
 
@@ -82,9 +91,9 @@ public:
         return name;
     }
 
-    static FunctionPtr create(ContextPtr)
+    static FunctionPtr create(ContextPtr context)
     {
-        return std::make_shared<ExtractKeyValuePairs>();
+        return std::make_shared<ExtractKeyValuePairs>(context);
     }
 
     ColumnPtr executeImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr &, size_t) const override
@@ -120,6 +129,9 @@ public:
     {
         return {1, 2, 3, 4};
     }
+
+private:
+    ContextPtr context;
 };
 
 struct NameExtractKeyValuePairs

src/Functions/keyvaluepair/impl/CHKeyValuePairExtractor.h

@@ -13,6 +13,7 @@ namespace DB
 namespace ErrorCodes
 {
     extern const int LOGICAL_ERROR;
+    extern const int LIMIT_EXCEEDED;
 }
 
 /*
@@ -25,8 +26,8 @@ class CHKeyValuePairExtractor : public KeyValuePairExtractor
     using NextState = DB::extractKV::StateHandler::NextState;
 
 public:
-    explicit CHKeyValuePairExtractor(StateHandler state_handler_)
-        : state_handler(std::move(state_handler_))
+    explicit CHKeyValuePairExtractor(StateHandler state_handler_, uint64_t max_number_of_pairs_)
+        : state_handler(std::move(state_handler_)), max_number_of_pairs(max_number_of_pairs_)
     {}
 
     uint64_t extract(const std::string & data, ColumnString::MutablePtr & keys, ColumnString::MutablePtr & values) override
@@ -113,11 +114,16 @@ private:
     NextState flushPair(const std::string_view & file, auto & key,
                         auto & value, uint64_t & row_offset)
     {
+        row_offset++;
+
+        if (row_offset > max_number_of_pairs)
+        {
+            throw Exception(ErrorCodes::LIMIT_EXCEEDED, "Number of pairs produced exceeded the limit of {}", max_number_of_pairs);
+        }
+
         key.commit();
         value.commit();
 
-        row_offset++;
-
         return {0, file.empty() ? State::END : State::WAITING_KEY};
     }
 
@@ -128,6 +134,7 @@ private:
     }
 
     StateHandler state_handler;
+    uint64_t max_number_of_pairs;
 };
 
 }

src/Functions/keyvaluepair/impl/KeyValuePairExtractorBuilder.cpp

@@ -31,6 +31,12 @@ KeyValuePairExtractorBuilder & KeyValuePairExtractorBuilder::withEscaping()
     return *this;
 }
 
+KeyValuePairExtractorBuilder & KeyValuePairExtractorBuilder::withMaxNumberOfPairs(uint64_t max_number_of_pairs_)
+{
+    max_number_of_pairs = max_number_of_pairs_;
+    return *this;
+}
+
 std::shared_ptr<KeyValuePairExtractor> KeyValuePairExtractorBuilder::build() const
 {
     if (with_escaping)
@@ -46,9 +52,9 @@ namespace
 using namespace extractKV;
 
 template <typename T>
-auto makeStateHandler(const T && handler)
+auto makeStateHandler(const T && handler, uint64_t max_number_of_pairs)
 {
-    return std::make_shared<CHKeyValuePairExtractor<T>>(handler);
+    return std::make_shared<CHKeyValuePairExtractor<T>>(handler, max_number_of_pairs);
 }
 
 }
@@ -57,14 +63,14 @@ std::shared_ptr<KeyValuePairExtractor> KeyValuePairExtractorBuilder::buildWithou
 {
     auto configuration = ConfigurationFactory::createWithoutEscaping(key_value_delimiter, quoting_character, item_delimiters);
 
-    return makeStateHandler(NoEscapingStateHandler(configuration));
+    return makeStateHandler(NoEscapingStateHandler(configuration), max_number_of_pairs);
 }
 
 std::shared_ptr<KeyValuePairExtractor> KeyValuePairExtractorBuilder::buildWithEscaping() const
 {
     auto configuration = ConfigurationFactory::createWithEscaping(key_value_delimiter, quoting_character, item_delimiters);
 
-    return makeStateHandler(InlineEscapingStateHandler(configuration));
+    return makeStateHandler(InlineEscapingStateHandler(configuration), max_number_of_pairs);
 }
 
 }

src/Functions/keyvaluepair/impl/KeyValuePairExtractorBuilder.h

@@ -20,6 +20,8 @@ public:
 
     KeyValuePairExtractorBuilder & withEscaping();
 
+    KeyValuePairExtractorBuilder & withMaxNumberOfPairs(uint64_t max_number_of_pairs_);
+
     std::shared_ptr<KeyValuePairExtractor> build() const;
 
 private:
@@ -27,6 +29,7 @@ private:
     char key_value_delimiter = ':';
     char quoting_character = '"';
     std::vector<char> item_delimiters = {' ', ',', ';'};
+    uint64_t max_number_of_pairs = std::numeric_limits<uint64_t>::max();
 
     std::shared_ptr<KeyValuePairExtractor> buildWithEscaping() const;
 

src/Interpreters/QueryLog.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <Common/ProfileEvents.h>
 #include <Core/NamesAndTypes.h>
 #include <Core/NamesAndAliases.h>
 #include <Core/Settings.h>

src/Interpreters/QueryViewsLog.h

@@ -13,6 +13,7 @@
 #include <Core/NamesAndAliases.h>
 #include <Interpreters/SystemLog.h>
 #include <base/types.h>
+#include <Common/ProfileEvents.h>
 
 namespace ProfileEvents
 {

src/Interpreters/tests/gtest_context_race.cpp

@@ -1,6 +1,7 @@
 #include <Interpreters/Context.h>
 #include <Common/tests/gtest_global_context.h>
 #include <gtest/gtest.h>
+#include <thread>
 
 using namespace DB;
 

src/Parsers/Access/ASTRowPolicyName.cpp

@@ -30,6 +30,11 @@ void ASTRowPolicyName::replaceEmptyDatabase(const String & current_database)
         full_name.database = current_database;
 }
 
+String ASTRowPolicyNames::tableOrAsterisk(const String & table_name) const
+{
+    return table_name == RowPolicyName::ANY_TABLE_MARK ? "*" : backQuoteIfNeed(table_name);
+}
+
 
 void ASTRowPolicyNames::formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const
 {
@@ -73,7 +78,7 @@ void ASTRowPolicyNames::formatImpl(const FormatSettings & settings, FormatState
             const String & table_name = full_name.table_name;
             if (!database.empty())
                 settings.ostr << backQuoteIfNeed(database) + ".";
-            settings.ostr << backQuoteIfNeed(table_name);
+            settings.ostr << tableOrAsterisk(table_name);
         }
     }
     else if (same_db_and_table_name)
@@ -92,7 +97,7 @@ void ASTRowPolicyNames::formatImpl(const FormatSettings & settings, FormatState
         settings.ostr << (settings.hilite ? hilite_keyword : "") << " ON " << (settings.hilite ? hilite_none : "");
         if (!database.empty())
             settings.ostr << backQuoteIfNeed(database) + ".";
-        settings.ostr << backQuoteIfNeed(table_name);
+        settings.ostr << tableOrAsterisk(table_name);
     }
     else
     {
@@ -108,7 +113,7 @@ void ASTRowPolicyNames::formatImpl(const FormatSettings & settings, FormatState
                           << (settings.hilite ? hilite_none : "");
             if (!database.empty())
                 settings.ostr << backQuoteIfNeed(database) + ".";
-            settings.ostr << backQuoteIfNeed(table_name);
+            settings.ostr << tableOrAsterisk(table_name);
         }
     }
 

src/Parsers/Access/ASTRowPolicyName.h

@@ -45,5 +45,8 @@ public:
     ASTPtr getRewrittenASTWithoutOnCluster(const WithoutOnClusterASTRewriteParams &) const override { return removeOnCluster<ASTRowPolicyNames>(clone()); }
 
     void replaceEmptyDatabase(const String & current_database);
+
+private:
+    String tableOrAsterisk(const String & table_name) const;
 };
 }

src/Parsers/Access/ParserRowPolicyName.cpp

@@ -26,8 +26,18 @@ namespace
         return IParserBase::wrapParseImpl(pos, [&]
         {
             String res_database, res_table_name;
-            if (!parseDatabaseAndTableName(pos, expected, res_database, res_table_name))
+            bool is_any_database = false;
+            bool is_any_table = false;
+
+            if (!parseDatabaseAndTableNameOrAsterisks(pos, expected, res_database, is_any_database, res_table_name, is_any_table)
+                || is_any_database)
+            {
                 return false;
+            }
+            else if (is_any_table)
+            {
+                res_table_name = RowPolicyName::ANY_TABLE_MARK;
+            }
 
             /// If table is specified without DB it cannot be followed by "ON"
             /// (but can be followed by "ON CLUSTER").

src/Processors/Formats/Impl/BSONEachRowRowInputFormat.cpp

@@ -64,20 +64,22 @@ inline size_t BSONEachRowRowInputFormat::columnIndex(const StringRef & name, siz
     /// Optimization by caching the order of fields (which is almost always the same)
     /// and a quick check to match the next expected field, instead of searching the hash table.
 
-    if (prev_positions.size() > key_index && prev_positions[key_index] && name == prev_positions[key_index]->getKey())
+    if (prev_positions.size() > key_index
+        && prev_positions[key_index] != Block::NameMap::const_iterator{}
+        && name == prev_positions[key_index]->first)
     {
-        return prev_positions[key_index]->getMapped();
+        return prev_positions[key_index]->second;
     }
     else
     {
-        auto * it = name_map.find(name);
+        const auto it = name_map.find(name);
 
-        if (it)
+        if (it != name_map.end())
         {
             if (key_index < prev_positions.size())
                 prev_positions[key_index] = it;
 
-            return it->getMapped();
+            return it->second;
         }
         else
             return UNKNOWN_FIELD;

src/Processors/Formats/Impl/BSONEachRowRowInputFormat.h

@@ -91,7 +91,7 @@ private:
     Block::NameMap name_map;
 
     /// Cached search results for previous row (keyed as index in JSON object) - used as a hint.
-    std::vector<Block::NameMap::LookupResult> prev_positions;
+    std::vector<Block::NameMap::const_iterator> prev_positions;
 
     DataTypes types;
 

src/Processors/Formats/Impl/JSONColumnsBlockInputFormatBase.cpp

@@ -128,7 +128,7 @@ Chunk JSONColumnsBlockInputFormatBase::generate()
         {
             /// Check if this name appears in header. If no, skip this column or throw
             /// an exception according to setting input_format_skip_unknown_fields
-            if (!name_to_index.has(*column_name))
+            if (name_to_index.find(*column_name) == name_to_index.end())
             {
                 if (!format_settings.skip_unknown_fields)
                     throw Exception(ErrorCodes::INCORRECT_DATA, "Unknown column found in input data: {}", *column_name);

src/Processors/Formats/Impl/JSONEachRowRowInputFormat.cpp

@@ -71,21 +71,20 @@ inline size_t JSONEachRowRowInputFormat::columnIndex(StringRef name, size_t key_
     /// and a quick check to match the next expected field, instead of searching the hash table.
 
     if (prev_positions.size() > key_index
-        && prev_positions[key_index]
-        && name == prev_positions[key_index]->getKey())
+        && prev_positions[key_index] != Block::NameMap::const_iterator{}
+        && name == prev_positions[key_index]->first)
     {
-        return prev_positions[key_index]->getMapped();
+        return prev_positions[key_index]->second;
     }
     else
     {
-        auto * it = name_map.find(name);
-
-        if (it)
+        const auto it = name_map.find(name);
+        if (it != name_map.end())
         {
             if (key_index < prev_positions.size())
                 prev_positions[key_index] = it;
 
-            return it->getMapped();
+            return it->second;
         }
         else
             return UNKNOWN_FIELD;

src/Processors/Formats/Impl/JSONEachRowRowInputFormat.h

@@ -71,11 +71,10 @@ private:
     /// for row like {..., "non-nullable column name" : null, ...}
 
     /// Hash table match `field name -> position in the block`. NOTE You can use perfect hash map.
-    using NameMap = HashMap<StringRef, size_t, StringRefHash>;
-    NameMap name_map;
+    Block::NameMap name_map;
 
     /// Cached search results for previous row (keyed as index in JSON object) - used as a hint.
-    std::vector<NameMap::LookupResult> prev_positions;
+    std::vector<Block::NameMap::const_iterator> prev_positions;
 
     bool allow_new_rows = true;
 

src/Processors/QueryPlan/DistributedCreateLocalPlan.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <Core/QueryProcessingStage.h>
+#include <Core/UUID.h>
 #include <Parsers/IAST_fwd.h>
 #include <Processors/QueryPlan/QueryPlan.h>
 #include <Processors/ResizeProcessor.h>

src/Processors/QueryPlan/Optimizations/projectionsCommon.cpp

@@ -42,6 +42,10 @@ bool canUseProjectionForReadingStep(ReadFromMergeTree * reading)
     if (reading->getContext()->getSettingsRef().allow_experimental_query_deduplication)
         return false;
 
+    // Currently projection don't support settings which implicitly modify aggregate functions.
+    if (reading->getContext()->getSettingsRef().aggregate_functions_null_for_empty)
+        return false;
+
     return true;
 }
 

src/Storages/DataLakes/DeltaLakeMetadataParser.cpp

@@ -11,10 +11,11 @@
 #include <Processors/Formats/Impl/ParquetBlockInputFormat.h>
 #include <Processors/Formats/Impl/ArrowColumnToCHColumn.h>
 #include <Formats/FormatFactory.h>
-#include <boost/algorithm/string/case_conv.hpp>
-#include <parquet/arrow/reader.h>
 #include <Columns/ColumnString.h>
 #include <Columns/ColumnNullable.h>
+#include <IO/ReadHelpers.h>
+#include <boost/algorithm/string/case_conv.hpp>
+#include <parquet/arrow/reader.h>
 #include <ranges>
 
 namespace fs = std::filesystem;

src/Storages/DataLakes/HudiMetadataParser.cpp

@@ -1,6 +1,7 @@
 #include <Storages/DataLakes/HudiMetadataParser.h>
 #include <Common/logger_useful.h>
 #include <ranges>
+#include <base/find_symbols.h>
 #include <Poco/String.h>
 #include "config.h"
 #include <filesystem>

src/Storages/DataLakes/IcebergMetadataParser.cpp

@@ -12,6 +12,7 @@
 #include <Storages/StorageS3.h>
 #include <Processors/Formats/Impl/AvroRowInputFormat.h>
 #include <Formats/FormatFactory.h>
+#include <IO/ReadHelpers.h>
 
 #include <Poco/JSON/Array.h>
 #include <Poco/JSON/Object.h>

src/Storages/StorageMongoDB.h

@@ -7,7 +7,7 @@
 namespace DB
 {
 /* Implements storage in the MongoDB database.
- * Use ENGINE = mysql(host_port, database_name, table_name, user_name, password)
+ * Use ENGINE = MongoDB(host:port, database, collection, user, password [, options]);
  * Read only.
  */
 

src/Storages/StorageS3.h

@@ -16,6 +16,7 @@
 #include <Poco/URI.h>
 #include <IO/S3/getObjectInfo.h>
 #include <IO/CompressionMethod.h>
+#include <IO/SeekableReadBuffer.h>
 #include <Interpreters/Context.h>
 #include <Interpreters/threadPoolCallbackRunner.h>
 #include <Storages/Cache/SchemaCache.h>

tests/broken_tests.txt

@@ -138,3 +138,4 @@
 01600_parts_states_metrics_long
 01600_parts_types_metrics_long
 01287_max_execution_speed
+02703_row_policy_for_database

tests/clickhouse-test

@@ -530,6 +530,7 @@ class SettingsRandomizer:
         "max_threads": lambda: random.randint(1, 64),
         "optimize_or_like_chain": lambda: random.randint(0, 1),
         "optimize_read_in_order": lambda: random.randint(0, 1),
+        "enable_multiple_prewhere_read_steps": lambda: random.randint(0, 1),
         "read_in_order_two_level_merge_threshold": lambda: random.randint(0, 100),
         "optimize_aggregation_in_order": lambda: random.randint(0, 1),
         "aggregation_in_order_max_block_bytes": lambda: random.randint(0, 50000000),

tests/queries/0_stateless/01710_projection_aggregate_functions_null_for_empty.reference

@@ -0,0 +1 @@
+1554690688

tests/queries/0_stateless/01710_projection_aggregate_functions_null_for_empty.sql

@@ -0,0 +1,8 @@
+DROP TABLE IF EXISTS t1;
+
+CREATE TABLE t1 (c0 Int32, PRIMARY KEY (c0)) ENGINE=MergeTree;
+INSERT INTO t1 VALUES (1554690688);
+
+SELECT MIN(t1.c0) FROM t1 SETTINGS aggregate_functions_null_for_empty = 1;
+
+DROP TABLE IF EXISTS t1;

tests/queries/0_stateless/02499_extract_key_value_pairs_multiple_input.reference

@@ -292,6 +292,35 @@ SELECT
     x;
 {'age':'31','last_key':'last_value','name':'neymar','nationality':'brazil','team':'psg'}
 -- { echoOn }
+
+SET extract_kvp_max_pairs_per_row = 2;
+-- Should be allowed because it no longer exceeds the max number of pairs
+-- expected output: {'key1':'value1','key2':'value2'}
+WITH
+    extractKeyValuePairs('key1:value1,key2:value2') AS s_map,
+    CAST(
+            arrayMap(
+                    (x) -> (x, s_map[x]), arraySort(mapKeys(s_map))
+                ),
+            'Map(String,String)'
+        ) AS x
+SELECT
+    x;
+{'key1':'value1','key2':'value2'}
+SET extract_kvp_max_pairs_per_row = 0;
+-- Should be allowed because max pairs per row is set to 0 (unlimited)
+-- expected output: {'key1':'value1','key2':'value2'}
+WITH
+    extractKeyValuePairs('key1:value1,key2:value2') AS s_map,
+    CAST(
+            arrayMap(
+                    (x) -> (x, s_map[x]), arraySort(mapKeys(s_map))
+                ),
+            'Map(String,String)'
+        ) AS x
+SELECT
+    x;
+{'key1':'value1','key2':'value2'}
 -- should not fail because pair delimiters contains 8 characters, which is within the limit
 WITH
     extractKeyValuePairs('not_important', ':', '12345678', '\'') AS s_map,

tests/queries/0_stateless/02499_extract_key_value_pairs_multiple_input.sql

@@ -414,7 +414,49 @@ WITH
 SELECT
     x; -- {serverError NUMBER_OF_ARGUMENTS_DOESNT_MATCH}
 
+-- Should fail allowed because it exceeds the max number of pairs
+SET extract_kvp_max_pairs_per_row = 1;
+WITH
+    extractKeyValuePairs('key1:value1,key2:value2') AS s_map,
+    CAST(
+            arrayMap(
+                    (x) -> (x, s_map[x]), arraySort(mapKeys(s_map))
+                ),
+            'Map(String,String)'
+        ) AS x
+SELECT
+    x; -- {serverError LIMIT_EXCEEDED}
+
 -- { echoOn }
+
+SET extract_kvp_max_pairs_per_row = 2;
+-- Should be allowed because it no longer exceeds the max number of pairs
+-- expected output: {'key1':'value1','key2':'value2'}
+WITH
+    extractKeyValuePairs('key1:value1,key2:value2') AS s_map,
+    CAST(
+            arrayMap(
+                    (x) -> (x, s_map[x]), arraySort(mapKeys(s_map))
+                ),
+            'Map(String,String)'
+        ) AS x
+SELECT
+    x;
+
+SET extract_kvp_max_pairs_per_row = 0;
+-- Should be allowed because max pairs per row is set to 0 (unlimited)
+-- expected output: {'key1':'value1','key2':'value2'}
+WITH
+    extractKeyValuePairs('key1:value1,key2:value2') AS s_map,
+    CAST(
+            arrayMap(
+                    (x) -> (x, s_map[x]), arraySort(mapKeys(s_map))
+                ),
+            'Map(String,String)'
+        ) AS x
+SELECT
+    x;
+
 -- should not fail because pair delimiters contains 8 characters, which is within the limit
 WITH
     extractKeyValuePairs('not_important', ':', '12345678', '\'') AS s_map,

tests/queries/0_stateless/02703_row_policies_for_asterisk.reference

@@ -0,0 +1,2 @@
+Policy for table `*` does not affect other tables in the database
+other	100	20

tests/queries/0_stateless/02703_row_policies_for_asterisk.sql

@@ -0,0 +1,11 @@
+-- Tags: no-parallel
+
+SELECT 'Policy for table `*` does not affect other tables in the database';
+CREATE DATABASE 02703_db_asterisk;
+CREATE ROW POLICY 02703_asterisk ON 02703_db_asterisk.`*` USING x=1 AS permissive TO ALL;
+CREATE TABLE 02703_db_asterisk.`*` (x UInt8, y UInt8) ENGINE = MergeTree ORDER BY x AS SELECT 100, 20;
+CREATE TABLE 02703_db_asterisk.`other` (x UInt8, y UInt8) ENGINE = MergeTree ORDER BY x AS SELECT 100, 20;
+SELECT 'star', * FROM 02703_db_asterisk.`*`;
+SELECT 'other', * FROM 02703_db_asterisk.other;
+DROP ROW POLICY 02703_asterisk ON 02703_db_asterisk.`*`;
+DROP DATABASE 02703_db_asterisk;

tests/queries/0_stateless/02703_row_policies_for_database_combination.reference

@@ -0,0 +1,42 @@
+None
+1	10
+2	20
+3	30
+4	40
+R1: x == 1
+1	10
+R1, R2: (x == 1) OR (x == 2)
+1	10
+2	20
+R1, R2: (x == 2) FROM ANOTHER
+2	20
+R1, R2, R3: (x == 1) OR (x == 2) OR (x == 3)
+1	10
+2	20
+3	30
+R1, R2, R3, R4: ((x == 1) OR (x == 2) OR (x == 3)) AND (x <= 2)
+1	10
+2	20
+R1, R2, R3, R4, R5: ((x == 1) OR (x == 2) OR (x == 3)) AND (x <= 2) AND (y >= 20)
+2	20
+2	20
+R1, R2, R3, R4, R5: (x == 2) AND (y >= 20) FROM AFTER_RP
+2	20
+R1, R2, R3, R4, R5: (x == 2) AND (y >= 20) FROM ANOTHER
+2	20
+R2, R3, R4, R5: ((x == 2) OR (x == 3)) AND (x <= 2) AND (y >= 20)
+2	20
+R3, R4, R5: (x == 3) AND (x <= 2) AND (y >= 20)
+R4, R5: (x <= 2) AND (y >= 20)
+2	20
+R5: (x >= 2)
+2	20
+3	30
+4	40
+Policy not applicable
+None
+1	10
+2	20
+3	30
+4	40
+No problematic policy, select works

tests/queries/0_stateless/02703_row_policies_for_database_combination.sql

@@ -0,0 +1,88 @@
+-- Tags: no-parallel
+
+DROP DATABASE IF EXISTS 02703_db;
+CREATE DATABASE 02703_db;
+DROP TABLE IF EXISTS 02703_db.02703_rptable;
+DROP TABLE IF EXISTS 02703_db.02703_rptable_another;
+CREATE TABLE 02703_db.02703_rptable (x UInt8, y UInt8) ENGINE = MergeTree ORDER BY x;
+
+INSERT INTO 02703_db.02703_rptable VALUES (1, 10), (2, 20), (3, 30), (4, 40);
+
+CREATE TABLE 02703_db.02703_rptable_another ENGINE = MergeTree ORDER BY x AS SELECT * FROM 02703_db.02703_rptable;
+
+
+DROP ROW POLICY IF EXISTS 02703_filter_1 ON 02703_db.02703_rptable;
+DROP ROW POLICY IF EXISTS 02703_filter_2 ON 02703_db.*;
+DROP ROW POLICY IF EXISTS 02703_filter_3 ON 02703_db.02703_rptable;
+DROP ROW POLICY IF EXISTS 02703_filter_4 ON 02703_db.02703_rptable;
+DROP ROW POLICY IF EXISTS 02703_filter_5 ON 02703_db.*;
+
+-- the test assumes users_without_row_policies_can_read_rows is true
+
+SELECT 'None';
+SELECT * FROM 02703_db.02703_rptable;
+
+CREATE ROW POLICY 02703_filter_1 ON 02703_db.02703_rptable USING x=1 AS permissive TO ALL;
+SELECT 'R1: x == 1';
+SELECT * FROM 02703_db.02703_rptable;
+
+CREATE ROW POLICY 02703_filter_2 ON 02703_db.* USING x=2 AS permissive TO ALL;
+SELECT 'R1, R2: (x == 1) OR (x == 2)';
+SELECT * FROM 02703_db.02703_rptable;
+
+SELECT 'R1, R2: (x == 2) FROM ANOTHER';
+SELECT * FROM 02703_db.02703_rptable_another;
+
+CREATE ROW POLICY 02703_filter_3 ON 02703_db.02703_rptable USING x=3 AS permissive TO ALL;
+SELECT 'R1, R2, R3: (x == 1) OR (x == 2) OR (x == 3)';
+SELECT * FROM 02703_db.02703_rptable;
+
+CREATE ROW POLICY 02703_filter_4 ON 02703_db.02703_rptable USING x<=2 AS restrictive TO ALL;
+SELECT 'R1, R2, R3, R4: ((x == 1) OR (x == 2) OR (x == 3)) AND (x <= 2)';
+SELECT * FROM 02703_db.02703_rptable;
+
+CREATE ROW POLICY 02703_filter_5 ON 02703_db.* USING y>=20 AS restrictive TO ALL;
+SELECT 'R1, R2, R3, R4, R5: ((x == 1) OR (x == 2) OR (x == 3)) AND (x <= 2) AND (y >= 20)';
+SELECT * FROM 02703_db.02703_rptable;
+
+CREATE TABLE 02703_db.02703_after_rp ENGINE = MergeTree ORDER BY x AS SELECT * FROM 02703_db.02703_rptable;
+SELECT * FROM 02703_db.02703_after_rp;
+
+-- does not matter if policies or table are created first
+SELECT 'R1, R2, R3, R4, R5: (x == 2) AND (y >= 20) FROM AFTER_RP';
+SELECT * FROM 02703_db.02703_after_rp;
+
+SELECT 'R1, R2, R3, R4, R5: (x == 2) AND (y >= 20) FROM ANOTHER';
+SELECT * FROM 02703_db.02703_rptable_another;
+
+DROP ROW POLICY 02703_filter_1 ON 02703_db.02703_rptable;
+SELECT 'R2, R3, R4, R5: ((x == 2) OR (x == 3)) AND (x <= 2) AND (y >= 20)';
+SELECT * FROM 02703_db.02703_rptable;
+
+DROP ROW POLICY 02703_filter_2 ON 02703_db.*;
+SELECT 'R3, R4, R5: (x == 3) AND (x <= 2) AND (y >= 20)';
+SELECT * FROM 02703_db.02703_rptable;
+
+DROP ROW POLICY 02703_filter_3 ON 02703_db.02703_rptable;
+SELECT 'R4, R5: (x <= 2) AND (y >= 20)';
+SELECT * FROM 02703_db.02703_rptable;
+
+DROP ROW POLICY 02703_filter_4 ON 02703_db.02703_rptable;
+SELECT 'R5: (x >= 2)';
+SELECT * FROM 02703_db.02703_rptable;
+
+CREATE TABLE 02703_db.02703_unexpected_columns (xx UInt8, yy UInt8) ENGINE = MergeTree ORDER BY xx;
+SELECT 'Policy not applicable';
+SELECT * FROM 02703_db.02703_unexpected_columns; -- { serverError 47 } -- Missing columns: 'x' while processing query
+
+DROP ROW POLICY 02703_filter_5 ON 02703_db.*;
+SELECT 'None';
+SELECT * FROM 02703_db.02703_rptable;
+
+SELECT 'No problematic policy, select works';
+SELECT 'Ok' FROM 02703_db.02703_unexpected_columns;
+
+DROP TABLE 02703_db.02703_rptable;
+DROP TABLE 02703_db.02703_rptable_another;
+DROP TABLE 02703_db.02703_unexpected_columns;
+DROP DATABASE 02703_db;

tests/queries/0_stateless/02703_row_policy_for_database.reference

@@ -0,0 +1,20 @@
+-- row policies for database
+  -- SHOW CREATE POLICY db1_02703 ON db1_02703.*
+CREATE ROW POLICY db1_02703 ON db1_02703.* FOR SELECT USING 1 TO ALL
+  -- SHOW CREATE POLICY ON db1_02703.*
+CREATE ROW POLICY db1_02703 ON db1_02703.* FOR SELECT USING 1 TO ALL
+CREATE ROW POLICY tbl1_02703 ON db1_02703.table FOR SELECT USING 1 TO ALL
+  -- SHOW CREATE POLICY ON db1_02703.`*`
+R1, R2: (x == 1) OR (x == 2)
+1
+2
+Check system.query_log
+SELECT \'-- row policies for database\';	[]
+SELECT \'  -- SHOW CREATE POLICY db1_02703 ON db1_02703.*\';	[]
+SELECT \'  -- SHOW CREATE POLICY ON db1_02703.*\';	[]
+SELECT \'  -- SHOW CREATE POLICY ON db1_02703.`*`\';	[]
+SELECT \'R1, R2: (x == 1) OR (x == 2)\';	[]
+SELECT * FROM 02703_rqtable_default;	['`02703_filter_11_db` ON default.*','`02703_filter_11` ON default.`02703_rqtable_default`']
+SELECT \'Check system.query_log\';	[]
+  -- CREATE DATABASE-LEVEL POLICY IN CURRENT DATABASE
+CREATE ROW POLICY db2_02703 ON db1_02703.* TO u1_02703

tests/queries/0_stateless/02703_row_policy_for_database.sql

@@ -0,0 +1,53 @@
+-- Tags: no-parallel
+
+DROP DATABASE IF EXISTS db1_02703;
+DROP USER IF EXISTS u1_02703;
+CREATE USER u1_02703;
+
+CREATE DATABASE db1_02703;
+
+CREATE TABLE db1_02703.02703_rqtable (x UInt8) ENGINE = MergeTree ORDER BY x;
+INSERT INTO db1_02703.02703_rqtable VALUES (1), (2), (3), (4);
+
+
+SELECT '-- row policies for database';
+CREATE ROW POLICY db1_02703 ON db1_02703.* USING 1 AS PERMISSIVE TO ALL;
+CREATE ROW POLICY tbl1_02703 ON db1_02703.table USING 1 AS PERMISSIVE TO ALL;
+SELECT '  -- SHOW CREATE POLICY db1_02703 ON db1_02703.*';
+SHOW CREATE POLICY db1_02703 ON db1_02703.*;
+SELECT '  -- SHOW CREATE POLICY ON db1_02703.*';
+SHOW CREATE POLICY ON db1_02703.*;
+SELECT '  -- SHOW CREATE POLICY ON db1_02703.`*`';
+SHOW CREATE POLICY ON db1_02703.`*`;
+DROP POLICY db1_02703 ON db1_02703.*;
+DROP POLICY tbl1_02703 ON db1_02703.table;
+
+CREATE ROW POLICY any_02703 ON *.some_table USING 1 AS PERMISSIVE TO ALL; -- { clientError 62 }
+
+CREATE TABLE 02703_rqtable_default (x UInt8) ENGINE = MergeTree ORDER BY x;
+
+CREATE ROW POLICY 02703_filter_11_db ON * USING x=1 AS permissive TO ALL;
+CREATE ROW POLICY 02703_filter_11 ON 02703_rqtable_default USING x=2 AS permissive TO ALL;
+
+INSERT INTO 02703_rqtable_default VALUES (1), (2), (3), (4);
+
+SELECT 'R1, R2: (x == 1) OR (x == 2)';
+SELECT * FROM 02703_rqtable_default;
+
+DROP TABLE 02703_rqtable_default;
+
+SELECT 'Check system.query_log';
+SYSTEM FLUSH LOGS;
+SELECT query, used_row_policies FROM system.query_log WHERE current_database == currentDatabase() AND type == 'QueryStart' AND query_kind == 'Select' ORDER BY event_time_microseconds;
+
+DROP ROW POLICY 02703_filter_11_db ON *;
+DROP ROW POLICY 02703_filter_11 ON 02703_rqtable_default;
+
+USE db1_02703;
+SELECT '  -- CREATE DATABASE-LEVEL POLICY IN CURRENT DATABASE';
+CREATE ROW POLICY db2_02703 ON * TO u1_02703;
+SHOW CREATE POLICY db2_02703 ON *;
+
+DROP ROW POLICY db2_02703 ON *;
+
+DROP USER u1_02703;

tests/queries/0_stateless/02731_parquet_s3.reference

@@ -0,0 +1 @@
+12639441726720293784

tests/queries/0_stateless/02731_parquet_s3.sql

@@ -0,0 +1,7 @@
+-- Tags: no-fasttest
+-- Tag no-fasttest: Depends on AWS
+
+-- Reading from s3 a parquet file of size between ~1 MB and ~2 MB was broken at some point.
+insert into function s3(s3_conn, filename='test_02731_parquet_s3.parquet') select cityHash64(number) from numbers(170000) settings s3_truncate_on_insert=1;
+
+select sum(*) from s3(s3_conn, filename='test_02731_parquet_s3.parquet') settings remote_filesystem_read_method='threadpool', remote_filesystem_read_prefetch=1;

tests/queries/0_stateless/02736_bit_count_big_int.reference

@@ -0,0 +1,13 @@
+128
+256
+128
+256
+127
+255
+126
+255
+64
+UInt8
+UInt16
+UInt8
+UInt16

tests/queries/0_stateless/02736_bit_count_big_int.sql

@@ -0,0 +1,19 @@
+SELECT bitCount(CAST(-1 AS UInt128));
+SELECT bitCount(CAST(-1 AS UInt256));
+
+SELECT bitCount(CAST(-1 AS Int128));
+SELECT bitCount(CAST(-1 AS Int256));
+
+SELECT bitCount(CAST(-1 AS UInt128) - 1);
+SELECT bitCount(CAST(-1 AS UInt256) - 2);
+
+SELECT bitCount(CAST(-1 AS Int128) - 3);
+SELECT bitCount(CAST(-1 AS Int256) - 4);
+
+SELECT bitCount(CAST(0xFFFFFFFFFFFFFFFF AS Int256));
+
+SELECT toTypeName(bitCount(1::UInt128));
+SELECT toTypeName(bitCount(1::UInt256));
+
+SELECT toTypeName(bitCount(1::Int128));
+SELECT toTypeName(bitCount(1::Int256));