diff --git a/include/cpphibernate/driver/mariadb/helper/context.h b/include/cpphibernate/driver/mariadb/helper/context.h
index 561bea6..c3551c1 100644
--- a/include/cpphibernate/driver/mariadb/helper/context.h
+++ b/include/cpphibernate/driver/mariadb/helper/context.h
@@ -73,18 +73,7 @@ beg_namespace_cpphibernate_driver_mariadb
 
     public:
         template<typename T>
-        inline decltype(auto) get() const
-        {
-            if (!data)
-                throw misc::hibernate_exception("no data assigned!");
-            auto type_id = misc::get_type_id(hana::type_c<mp::decay_t<T>>);
-            if (type_id != data_id)
-            {
-                throw misc::hibernate_exception(static_cast<std::ostringstream&>(std::ostringstream { }
-                    << "invalid type! expected " << data_id << ", got " << type_id).str());
-            }
-            return *static_cast<T*>(data);
-        }
+        inline decltype(auto) get(const table_t* table = nullptr) const;
 
         template<typename T_data>
         inline data_context(
@@ -141,5 +130,46 @@ beg_namespace_cpphibernate_driver_mariadb
             { }
     };
 
+    /* read_context */
+
+    struct read_context
+        : public filter_context
+    {
+        bool        is_dynamic;
+        size_t      base_dataset_id;
+        std::string where;
+        std::string limit;
+        std::string order_by;
+
+        template<typename T_data>
+        inline read_context(
+                T_data&                     p_data,
+                const schema_t&             p_schema,
+                ::cppmariadb::connection&   p_connection,
+                const filter_t&             p_filter)
+            : filter_context    (p_data, p_schema, p_connection, p_filter)
+            , is_dynamic        (false)
+            , base_dataset_id   (0)
+            { }
+
+        virtual ~read_context() = default;
+
+        template<typename T_dataset>
+        inline T_dataset& emplace(const table_t* table = nullptr) const;
+
+        void emplace() const
+            { emplace_intern(nullptr); }
+
+        void finish() const
+            { finish_intern(); }
+
+    private:
+        virtual void* emplace_intern(void* data) const
+            { throw misc::hibernate_exception("read_context::emplace_intern is not implemented!"); }
+
+        virtual void finish_intern() const
+            { throw misc::hibernate_exception("read_context::finish_intern is not implemented!"); }
+    };
+
 }
 end_namespace_cpphibernate_driver_mariadb
\ No newline at end of file
diff --git a/include/cpphibernate/driver/mariadb/helper/context.inl b/include/cpphibernate/driver/mariadb/helper/context.inl
new file mode 100644
index 0000000..af6ac96
--- /dev/null
+++ b/include/cpphibernate/driver/mariadb/helper/context.inl
@@ -0,0 +1,67 @@
+#pragma once
+
+#include <cpphibernate/driver/mariadb/helper/context.h>
+
+beg_namespace_cpphibernate_driver_mariadb
+{
+
+    /* data_context */
+
+    template<typename T>
+    inline decltype(auto) data_context
+        ::get(const table_t* table) const
+    {
+        if (!data)
+            throw misc::hibernate_exception("no data assigned!");
+
+        auto type_id = misc::get_type_id(hana::type_c<mp::decay_t<T>>);
+        if (type_id != data_id)
+        {
+            if (!table)
+                table = &schema.table(type_id);
+
+            while(table && table->dataset_id != type_id)
+                table = table->base_table;
+
+            if (!table)
+            {
+                throw misc::hibernate_exception(static_cast<std::ostringstream&>(std::ostringstream { }
+                    << "invalid type! expected " << data_id << ", got " << type_id).str());
+            }
+        }
+        return *static_cast<T*>(data);
+    }
+
+    /* read_context */
+
+    template<typename T_dataset>
+    T_dataset& read_context
+        ::emplace(const table_t* table) const
+    {
+        if (!is_dynamic || base_dataset_id == 0)
+            throw misc::hibernate_exception("dynamic creation is deactivated for this context!");
+
+        // check table
+        auto dataset_id = misc::get_type_id(hana::type_c<mp::decay_t<T_dataset>>);
+        if (!table)
+            table = &schema.table(dataset_id);
+        else if (table->dataset_id != dataset_id)
+            throw misc::hibernate_exception("dataset id of table and dataset to insert defer!");
+
+        // check base
+        auto tbl = table;
+        while (tbl && tbl->dataset_id != base_dataset_id)
+            tbl = tbl->base_table;
+        if (!tbl)
+            throw misc::hibernate_exception(utl::type_helper<T_dataset>::name() + " is not a derived type of table " + table->table_name);
+
+        // create dataset
+        auto ptr = std::make_unique<T_dataset>();
+        auto ret = emplace_intern(ptr.get());
+        if (!ret)
+            ret = ptr.release();
+        return *static_cast<T_dataset*>(ret);
+    }
+
+}
+end_namespace_cpphibernate_driver_mariadb
\ No newline at end of file
diff --git a/include/cpphibernate/driver/mariadb/impl.h b/include/cpphibernate/driver/mariadb/impl.h
index b990d17..611d188 100644
--- a/include/cpphibernate/driver/mariadb/impl.h
+++ b/include/cpphibernate/driver/mariadb/impl.h
@@ -2,4 +2,8 @@
 
 #include <cpphibernate/driver/mariadb/impl/create_update.h>
 #include <cpphibernate/driver/mariadb/impl/limit.h>
-#include <cpphibernate/driver/mariadb/impl/where.h>
\ No newline at end of file
+#include <cpphibernate/driver/mariadb/impl/order_by.h>
+#include <cpphibernate/driver/mariadb/impl/read.h>
+#include <cpphibernate/driver/mariadb/impl/where.h>
+
+#include <cpphibernate/driver/mariadb/helper/context.inl>
\ No newline at end of file
diff --git a/include/cpphibernate/driver/mariadb/impl/limit.h b/include/cpphibernate/driver/mariadb/impl/limit.h
index 7e502c6..594f973 100644
--- a/include/cpphibernate/driver/mariadb/impl/limit.h
+++ b/include/cpphibernate/driver/mariadb/impl/limit.h
@@ -22,17 +22,17 @@ beg_namespace_cpphibernate_driver_mariadb
             hana::for_each(modifier, [&limit, &offset](auto& modifier){
                 using modifier_type  = mp::decay_t<decltype(modifier)>;
                 using is_limit_type  = modifier::is_limit_modifier<modifier_type>;
-                using is_offset_type = modifier::is_offset_modifier<modifier_type>;
+                using is_offset_type = modifier::is_offset<modifier_type>;
                 hana::eval_if(
                     is_limit_type { },
                     [&limit, &modifier](auto _){
-                        limit = hana::value(_(modifier).value);
+                        limit = static_cast<ssize_t>(hana::value(_(modifier).value));
                     },
                     [&offset, &modifier](){
                         hana::eval_if(
                             is_offset_type { },
                             [&offset, &modifier](auto _){
-                                offset = hana::value(_(modifier).value);
+                                offset = static_cast<ssize_t>(hana::value(_(modifier).value));
                             },
                             []{
                                 /* no-op */
@@ -48,7 +48,7 @@ beg_namespace_cpphibernate_driver_mariadb
                 std::ostringstream ss;
                 ss << "LIMIT " << limit;
                 if (offset >= 0)
-                    ss << " OFFSET" << offset;
+                    ss << " OFFSET " << offset;
                 statement.assign(ss.str());
             }
         }
diff --git a/include/cpphibernate/driver/mariadb/impl/modifier_tags.h b/include/cpphibernate/driver/mariadb/impl/modifier_tags.h
index f1c6a9b..020a4ff 100644
--- a/include/cpphibernate/driver/mariadb/impl/modifier_tags.h
+++ b/include/cpphibernate/driver/mariadb/impl/modifier_tags.h
@@ -24,7 +24,7 @@ beg_namespace_cpphibernate_driver_mariadb
             mp::list<T_modifier>,
             mp::enable_if_c<
                     modifier::is_limit_modifier<mp::decay_t<T_modifier>>::value
-                ||  modifier::is_offset_modifier<mp::decay_t<T_modifier>>::value>>
+                ||  modifier::is_offset<mp::decay_t<T_modifier>>::value>>
         {
             static constexpr decltype(auto) apply(T_modifier&&)
                 { return T_modifier { }; }
diff --git a/include/cpphibernate/driver/mariadb/impl/order_by.h b/include/cpphibernate/driver/mariadb/impl/order_by.h
new file mode 100644
index 0000000..cd85783
--- /dev/null
+++ b/include/cpphibernate/driver/mariadb/impl/order_by.h
@@ -0,0 +1,83 @@
+#pragma once
+
+#include <sstream>
+#include <cpphibernate/config.h>
+#include <cpphibernate/modifier.h>
+#include <cpphibernate/driver/mariadb/schema/schema.h>
+
+beg_namespace_cpphibernate_driver_mariadb
+{
+
+    /* order_by_builder */
+
+    template<typename T_modifiers>
+    struct order_by_builder
+    {
+    private:
+        struct build_t
+        {
+            const schema_t&     schema;
+            const T_modifiers&  modifiers;
+            std::ostringstream  os;
+
+            inline build_t(const schema_t& p_schema, const T_modifiers& p_modifiers)
+                : schema    (p_schema)
+                , modifiers (p_modifiers)
+                { }
+
+            inline void build(::cppmariadb::statement& statement)
+            {
+                size_t index = 0;
+                hana::for_each(modifiers, [&](auto& modifier){
+                    using modifier_type    = mp::decay_t<decltype(modifier)>;
+                    using is_order_by_type = modifier::is_order_by<modifier_type>;
+                    hana::eval_if(
+                        is_order_by_type { },
+                        [&](auto _){
+                            hana::for_each(_(modifier).fields, [&](auto& f){
+                                using field_type        = mp::decay_t<decltype(f)>;
+                                using is_ascencing_type = modifier::is_ascending<field_type>;
+                                if (index++ == 0)   os << "ORDER_BY ";
+                                else                os << ", ";
+                                auto& field = schema.field(misc::get_type_id(f.wrapped_field));
+                                os  <<  "`"
+                                    <<  field.table_name
+                                    <<  "`.`"
+                                    <<  field.field_name
+                                    <<  "` "
+                                    <<  (is_ascencing_type::value
+                                            ? "ASC"
+                                            : "DESC");
+                            });
+                        },
+                        []{ });
+                });
+                statement.assign(os.str());
+            }
+        };
+
+    private:
+        const schema_t*         _schema     { nullptr };
+        ::cppmariadb::statement _statement;
+
+    public:
+        inline ::cppmariadb::statement& assign(const schema_t& schema, const T_modifiers& modifiers)
+        {
+            if (_schema != &schema)
+            {
+                build_t(schema, modifiers).build(_statement);
+                _schema = &schema;
+            }
+            return _statement;
+        }
+    };
+
+    template<typename T_modifiers>
+    inline decltype(auto) build_order_by(const schema_t& schema, const T_modifiers& modifiers)
+    {
+        static order_by_builder<T_modifiers> builder;
+        return builder.assign(schema, modifiers);
+    }
+
+}
+end_namespace_cpphibernate_driver_mariadb
diff --git a/include/cpphibernate/driver/mariadb/impl/read.h b/include/cpphibernate/driver/mariadb/impl/read.h
new file mode 100644
index 0000000..bd8143c
--- /dev/null
+++ b/include/cpphibernate/driver/mariadb/impl/read.h
@@ -0,0 +1,93 @@
+#pragma once
+
+#include <cppmariadb.h>
+#include <cpphibernate/misc.h>
+#include <cpphibernate/config.h>
+#include <cpphibernate/driver/mariadb/helper.h>
+#include <cpphibernate/driver/mariadb/schema/schema.h>
+
+beg_namespace_cpphibernate_driver_mariadb
+{
+
+    /* read_impl_t */
+
+    template<typename T_dataset, typename = void>
+    struct read_impl_t
+    {
+        using dataset_type = T_dataset;
+
+        struct context_impl
+            : public read_context
+        {
+            mutable size_t count;
+
+            template<typename T_read_context>
+            context_impl(T_read_context&& p_read_context)
+                : read_context  (std::forward<T_read_context>(p_read_context))
+                , count         (0)
+                { }
+
+        private:
+            virtual void* emplace_intern(void* data) const override
+            {
+                if (data)
+                    throw misc::hibernate_exception("This context has a pre assigned dataset and can therefor not work in dynamic mode!");
+                ++count;
+                if (count > 1)
+                    throw misc::hibernate_exception("Expected exactly one dataset, but received more!");
+                return nullptr;
+            }
+
+            virtual void finish_intern() const override
+            {
+                if (count < 1)
+                    throw misc::hibernate_exception("Expected exactly one dataset, but received none!");
+            }
+        };
+
+        static inline void apply(const read_context& context)
+        {
+            auto  dataset_id = misc::get_type_id(hana::type_c<dataset_type>);
+            auto& connection = context.connection;
+            auto& schema     = context.schema;
+            auto& table      = schema.table(dataset_id);
+
+            transaction_lock trans(connection);
+            table.read(context_impl(context));
+            trans.commit();
+        }
+    };
+
+    /* read_impl_t - nullable */
+
+    template<typename T_dataset>
+    struct read_impl_t<
+        T_dataset,
+        mp::enable_if<misc::is_nullable<T_dataset>>>
+    {
+        using dataset_type          = T_dataset;
+        using nullable_helper_type  = misc::nullable_helper<dataset_type>;
+
+        static inline void apply(const read_context& context)
+        {
+
+        }
+    };
+
+    /* read_impl_t - container */
+
+    template<typename T_dataset>
+    struct read_impl_t<
+        T_dataset,
+        mp::enable_if<misc::is_container<T_dataset>>>
+    {
+        using dataset_type = T_dataset;
+
+        static inline void apply(const read_context& context)
+        {
+
+        }
+    };
+
+}
+end_namespace_cpphibernate_driver_mariadb
\ No newline at end of file
diff --git a/include/cpphibernate/driver/mariadb/impl/where.h b/include/cpphibernate/driver/mariadb/impl/where.h
index 324a4b5..9af2d15 100644
--- a/include/cpphibernate/driver/mariadb/impl/where.h
+++ b/include/cpphibernate/driver/mariadb/impl/where.h
@@ -87,7 +87,7 @@ beg_namespace_cpphibernate_driver_mariadb
                 size_t index = 0;
                 hana::for_each(modifiers, [&](auto& modifier){
                     using modifier_type = mp::decay_t<decltype(modifier)>;
-                    using is_where_type = modifier::is_where_modifier<modifier_type>;
+                    using is_where_type = modifier::is_where<modifier_type>;
                     hana::eval_if(
                         is_where_type { },
                         [&](auto _){
@@ -144,7 +144,7 @@ beg_namespace_cpphibernate_driver_mariadb
             {
                 hana::for_each(modifiers, [&](auto& modifier){
                     using modifier_type = mp::decay_t<decltype(modifier)>;
-                    using is_where_type = modifier::is_where_modifier<modifier_type>;
+                    using is_where_type = modifier::is_where<modifier_type>;
                     hana::eval_if(
                         is_where_type { },
                         [&](auto _){
diff --git a/include/cpphibernate/driver/mariadb/mariadb.h b/include/cpphibernate/driver/mariadb/mariadb.h
index 2e7fe06..f84fd27 100644
--- a/include/cpphibernate/driver/mariadb/mariadb.h
+++ b/include/cpphibernate/driver/mariadb/mariadb.h
@@ -44,10 +44,11 @@ beg_namespace_cpphibernate_driver_mariadb
         template<typename T_dataset, typename T_modifiers>
         inline void read_impl(T_dataset& dataset, T_modifiers&& modifiers) const
         {
-            auto& where = build_where(_schema, modifiers);
-            auto& limit = build_limit(modifiers);
-            std::cout << "WHERE = " << where.query(_connection) << std::endl;
-            std::cout << "LIMIT = " << limit.query(_connection) << std::endl;
+            read_context context(dataset, _schema, _connection, _filter);
+            context.where    = build_where(_schema, modifiers).query(_connection);
+            context.limit    = build_limit(modifiers).query(_connection);
+            context.order_by = build_order_by(_schema, modifiers).query(_connection);
+            read_impl_t<T_dataset>::apply(context);
         }
     };
 
diff --git a/include/cpphibernate/driver/mariadb/schema/field.inl b/include/cpphibernate/driver/mariadb/schema/field.inl
index 1901a72..dd66fc4 100644
--- a/include/cpphibernate/driver/mariadb/schema/field.inl
+++ b/include/cpphibernate/driver/mariadb/schema/field.inl
@@ -39,7 +39,7 @@ beg_namespace_cpphibernate_driver_mariadb
     value_t value_field_t<T_field>
         ::get(const data_context& context) const
     {
-        auto& dataset = context.get<dataset_type>();
+        auto& dataset = context.get<dataset_type>(this->table);
         return type_props::convert_from(this->field.getter(dataset));
     }
 
@@ -47,7 +47,7 @@ beg_namespace_cpphibernate_driver_mariadb
     void value_field_t<T_field>
         ::set(const data_context& context, const value_t& value) const
     {
-        auto& dataset = context.get<dataset_type>();
+        auto& dataset = context.get<dataset_type>(this->table);
         this->field.setter(dataset, type_props::convert_to(value));
     }
 
diff --git a/include/cpphibernate/driver/mariadb/schema/filter.h b/include/cpphibernate/driver/mariadb/schema/filter.h
index 33d5fca..fae0931 100644
--- a/include/cpphibernate/driver/mariadb/schema/filter.h
+++ b/include/cpphibernate/driver/mariadb/schema/filter.h
@@ -16,12 +16,15 @@ beg_namespace_cpphibernate_driver_mariadb
         using field_set_type = std::set<const field_t*>;
         using table_set_type = std::set<const table_t*>;
 
-        size_t          cache_id;
+        size_t          cache_id { 0 };
         field_set_type  fields;
         table_set_type  tables;
 
-        bool contains(const table_t* table, bool check_base) const;
-        bool contains(const field_t* field) const;
+        inline bool is_excluded(const table_t& table) const
+            { return tables.count(&table); }
+
+        inline bool is_excluded(const field_t& field) const
+            { return fields.count(&field); }
     };
 
 }
diff --git a/include/cpphibernate/driver/mariadb/schema/table.h b/include/cpphibernate/driver/mariadb/schema/table.h
index 3a4c83c..ac5aa5d 100644
--- a/include/cpphibernate/driver/mariadb/schema/table.h
+++ b/include/cpphibernate/driver/mariadb/schema/table.h
@@ -76,8 +76,8 @@ beg_namespace_cpphibernate_driver_mariadb
         inline decltype(auto) create_update(const create_update_context& context) const
             { return create_update_intern(context); }
 
-        inline void read() const
-            {  }
+        inline void read(const read_context& context) const
+            { return read_exec(context); }
 
     private:
         template<typename T_schema, typename T_table, typename T_base_dataset>
@@ -87,14 +87,18 @@ beg_namespace_cpphibernate_driver_mariadb
         friend struct table_polymorphic_t;
 
         using statement_ptr = std::unique_ptr<::cppmariadb::statement>;
+        using statement_map = std::map<size_t, ::cppmariadb::statement>;
 
         mutable statement_ptr _statement_create_table;
         mutable statement_ptr _statement_alter_table;
         mutable statement_ptr _statement_insert_into;
+        mutable statement_map _statement_select_static;
+        mutable statement_map _statement_select_dynamic;
 
         ::cppmariadb::statement& get_statement_create_table() const;
         ::cppmariadb::statement* get_statement_alter_table() const;
         ::cppmariadb::statement& get_statement_insert_into() const;
+        ::cppmariadb::statement& get_statement_select(const read_context& context) const;
 
         std::string execute_create_update(
             const create_update_context&    context,
@@ -109,6 +113,8 @@ beg_namespace_cpphibernate_driver_mariadb
 
         virtual std::string create_update_intern(const create_update_context& context) const;
                 std::string create_update_exec  (const create_update_context& context) const;
+
+                void        read_exec           (const read_context& context) const;
     };
 
     /* table_simple_t */
diff --git a/include/cpphibernate/misc/container_helper.h b/include/cpphibernate/misc/container_helper.h
index 7f64153..0adc64d 100644
--- a/include/cpphibernate/misc/container_helper.h
+++ b/include/cpphibernate/misc/container_helper.h
@@ -12,10 +12,40 @@ beg_namespace_cpphibernate_misc
     /* container_helper */
 
     template<typename T_container, typename = void>
-    struct container_helper
+    struct container_helper;
+
+    template<typename T_value, typename... T_args>
+    struct container_helper<std::vector<T_value, T_args...>, void>
+    {
+        using container_type = std::vector<T_value, T_args...>;
+        using value_type     = T_value;
+
+        template<typename... T_xargs>
+        static inline value_type& emplace(container_type& container, T_xargs&&... args)
+        {
+            container.emplace_back(std::forward<T_xargs>(args)...);
+            return container.back();
+        }
+
+        static inline void clear(container_type& container)
+            { container.clear(); }
+    };
+
+    template<typename T_value, typename... T_args>
+    struct container_helper<std::list<T_value, T_args...>, void>
     {
-        using container_type = T_container;
-        using value_type     = real_dataset_t<container_type>;
+        using container_type = std::list<T_value, T_args...>;
+        using value_type     = T_value;
+
+        template<typename... T_xargs>
+        static inline value_type& emplace(container_type& container, T_xargs&&... args)
+        {
+            container.emplace_back(std::forward<T_xargs>(args)...);
+            return container.back();
+        }
+
+        static inline void clear(container_type& container)
+            { container.clear(); }
     };
 
 }
diff --git a/include/cpphibernate/modifier.h b/include/cpphibernate/modifier.h
index 474cf89..36dfe11 100644
--- a/include/cpphibernate/modifier.h
+++ b/include/cpphibernate/modifier.h
@@ -4,4 +4,5 @@
 #include <cpphibernate/modifier/modifier.h>
 #include <cpphibernate/modifier/modifiers.h>
 #include <cpphibernate/modifier/offset.h>
+#include <cpphibernate/modifier/order_by.h>
 #include <cpphibernate/modifier/where.h>
\ No newline at end of file
diff --git a/include/cpphibernate/modifier/limit.h b/include/cpphibernate/modifier/limit.h
index 0889090..39c3256 100644
--- a/include/cpphibernate/modifier/limit.h
+++ b/include/cpphibernate/modifier/limit.h
@@ -19,32 +19,6 @@ beg_namespace_cpphibernate_modifier
             using value_type = T_value;
 
             value_type value;
-
-            constexpr limit_t(T_value&& p_value)
-                : value(std::forward<T_value>(p_value))
-                { }
-        };
-
-        /* limit_builder */
-
-        template<typename X, typename = void>
-        struct limit_builder
-        {
-            template<typename... T_args>
-            static constexpr decltype(auto) apply(T_args&... args)
-                { static_assert(sizeof...(args) == -1, "Invalid parameters for hibernate::modifier::limit(...)!"); }
-        };
-
-        template<typename T_value>
-        struct limit_builder<
-            mp::list<T_value>,
-            mp::enable_if<mp::is_integral<T_value>>>
-        {
-            static constexpr decltype(auto) apply(T_value&& value)
-            {
-                using value_type = mp::integral_constant<T_value, value>;
-                return limit_t<value_type>(value_type { });
-            }
         };
 
     }
@@ -58,7 +32,8 @@ beg_namespace_cpphibernate_modifier
 
     /* make */
 
-    constexpr decltype(auto) limit = misc::make_generic_predicate<__impl::limit_builder> { };
+    template<size_t T_value>
+    constexpr decltype(auto) limit = __impl::limit_t<hana::size_t<T_value>> { };
 
 }
 end_namespace_cpphibernate_modifier
\ No newline at end of file
diff --git a/include/cpphibernate/modifier/offset.h b/include/cpphibernate/modifier/offset.h
index e1e49d1..20bbcec 100644
--- a/include/cpphibernate/modifier/offset.h
+++ b/include/cpphibernate/modifier/offset.h
@@ -19,32 +19,6 @@ beg_namespace_cpphibernate_modifier
             using value_type = T_value;
 
             value_type value;
-
-            constexpr offset_t(T_value&& p_value)
-                : value(std::forward<T_value>(p_value))
-                { }
-        };
-
-        /* offset_builder */
-
-        template<typename X, typename = void>
-        struct offset_builder
-        {
-            template<typename... T_args>
-            static constexpr decltype(auto) apply(T_args&... args)
-                { static_assert(sizeof...(args) == -1, "Invalid parameters for hibernate::modifier::offset(...)!"); }
-        };
-
-        template<typename T_value>
-        struct offset_builder<
-            mp::list<T_value>,
-            mp::enable_if<mp::is_integral<T_value>>>
-        {
-            static constexpr decltype(auto) apply(T_value&& value)
-            {
-                using value_type = mp::integral_constant<T_value, value>;
-                return offset_t<value_type>(value_type { });
-            }
         };
 
     }
@@ -52,13 +26,14 @@ beg_namespace_cpphibernate_modifier
     /* meta */
 
     template<typename T>
-    struct is_offset_modifier
+    struct is_offset
         : public mp::is_specialization_of<T, __impl::offset_t>
         { };
 
     /* make */
 
-    constexpr decltype(auto) offset = misc::make_generic_predicate<__impl::offset_builder> { };
+    template<size_t T_value>
+    constexpr decltype(auto) offset = __impl::offset_t<hana::size_t<T_value>> { };
 
 }
 end_namespace_cpphibernate_modifier
\ No newline at end of file
diff --git a/include/cpphibernate/modifier/order_by.h b/include/cpphibernate/modifier/order_by.h
new file mode 100644
index 0000000..515c0b9
--- /dev/null
+++ b/include/cpphibernate/modifier/order_by.h
@@ -0,0 +1,171 @@
+#pragma once
+
+#include <cpphibernate/misc.h>
+#include <cpphibernate/config.h>
+#include <cpphibernate/modifier/modifier.h>
+
+beg_namespace_cpphibernate_modifier
+{
+
+    namespace __impl
+    {
+
+        /* order_direction_tag_t */
+
+        struct order_direction_tag_t
+            { };
+
+        /* order_direction_t */
+
+        template<typename T_field>
+        struct order_direction_t
+            : public order_direction_tag_t
+        {
+            using field_type            = T_field;
+            using wrapped_field_type    = hana::type<mp::decay_t<field_type>>;
+
+            wrapped_field_type wrapped_field;
+        };
+
+    }
+
+    /* meta */
+
+    template<typename T>
+    struct is_order_direction
+        : public mp::is_base_of<__impl::order_direction_tag_t, T>
+        { };
+
+    template<typename... T>
+    struct all_are_order_directions
+        : public mp::all_true<is_order_direction<T>::value...>
+        { };
+
+    namespace __impl
+    {
+
+        /* order_ascending_t */
+
+        template<typename T_field>
+        struct order_ascending_t
+            : public order_direction_t<T_field>
+            { };
+
+        /* order_descending_t */
+
+        template<typename T_field>
+        struct order_descending_t
+            : public order_direction_t<T_field>
+            { };
+
+        /* order_by_t */
+
+        template<typename... T_fields>
+        struct order_by_t
+            : public modifier_t
+        {
+            using fields_type = hana::basic_tuple<T_fields...>;
+
+            fields_type fields;
+        };
+
+    }
+
+    /* meta */
+
+    template<typename T>
+    struct is_order_by
+        : public mp::is_specialization_of<T, __impl::order_by_t>
+        { };
+
+    template<typename T>
+    struct is_ascending
+        : public mp::is_specialization_of<T, __impl::order_ascending_t>
+        { };
+
+    template<typename T>
+    struct is_descending
+        : public mp::is_specialization_of<T, __impl::order_descending_t>
+        { };
+
+    namespace __impl
+    {
+
+        /* ascending_builder */
+
+        template<typename X, typename = void>
+        struct ascending_builder
+        {
+            template<typename... T_args>
+            static constexpr decltype(auto) apply(T_args&&... args)
+                { static_assert(sizeof...(args) == -1, "Invalid parameters for hibernate::modifier::ascending(...)!"); }
+        };
+
+        template<typename T_field>
+        struct ascending_builder<
+            mp::list<T_field>,
+            mp::enable_if<schema::is_field<mp::decay_t<T_field>>>>
+        {
+            static constexpr decltype(auto) apply(T_field&&)
+            {
+                using field_type     = mp::decay_t<T_field>;
+                using ascending_type = order_ascending_t<field_type>;
+                return ascending_type { };
+            }
+        };
+
+        /* descending_builder */
+
+        template<typename X, typename = void>
+        struct descending_builder
+        {
+            template<typename... T_args>
+            static constexpr decltype(auto) apply(T_args&&... args)
+                { static_assert(sizeof...(args) == -1, "Invalid parameters for hibernate::modifier::descending(...)!"); }
+        };
+
+        template<typename T_field>
+        struct descending_builder<
+            mp::list<T_field>,
+            mp::enable_if<schema::is_field<mp::decay_t<T_field>>>>
+        {
+            static constexpr decltype(auto) apply(T_field&&)
+            {
+                using field_type      = mp::decay_t<T_field>;
+                using descending_type = order_descending_t<field_type>;
+                return descending_type { };
+            }
+        };
+
+        /* order_by_builder */
+
+        template<typename X, typename = void>
+        struct order_by_builder
+        {
+            template<typename... T_args>
+            static constexpr decltype(auto) apply(T_args&&... args)
+                { static_assert(sizeof...(args) == -1, "Invalid parameters for hibernate::modifier::order_by(...)!"); }
+        };
+
+        template<typename... T_order_directions>
+        struct order_by_builder<
+            mp::list<T_order_directions...>,
+            mp::enable_if<all_are_order_directions<mp::decay_t<T_order_directions>...>>>
+        {
+            static constexpr decltype(auto) apply(T_order_directions&&...)
+            {
+                using order_by_type = order_by_t<mp::decay_t<T_order_directions>...>;
+                return order_by_type { };
+            }
+        };
+
+    }
+
+    /* make */
+
+    constexpr decltype(auto) ascending  = misc::make_generic_predicate<__impl::ascending_builder>  { };
+    constexpr decltype(auto) descending = misc::make_generic_predicate<__impl::descending_builder> { };
+    constexpr decltype(auto) order_by   = misc::make_generic_predicate<__impl::order_by_builder>   { };
+
+}
+end_namespace_cpphibernate_modifier
\ No newline at end of file
diff --git a/include/cpphibernate/modifier/where/where.h b/include/cpphibernate/modifier/where/where.h
index 16c33c6..685e24a 100644
--- a/include/cpphibernate/modifier/where/where.h
+++ b/include/cpphibernate/modifier/where/where.h
@@ -32,7 +32,7 @@ beg_namespace_cpphibernate_modifier
         struct where_builder
         {
             template<typename... T_args>
-            static constexpr decltype(auto) apply(T_args&... args)
+            static constexpr decltype(auto) apply(T_args&&... args)
                 { static_assert(sizeof...(args) == -1, "Invalid parameters for hibernate::modifier::where(...)!"); }
         };
 
@@ -48,7 +48,7 @@ beg_namespace_cpphibernate_modifier
     /* meta */
 
     template<typename T>
-    struct is_where_modifier
+    struct is_where
         : public mp::is_specialization_of<T, __impl::where_t>
         { };
 
diff --git a/src/driver/mariadb/schema/filter.cpp b/src/driver/mariadb/schema/filter.cpp
deleted file mode 100644
index 8fa2d5d..0000000
--- a/src/driver/mariadb/schema/filter.cpp
+++ /dev/null
@@ -1,19 +0,0 @@
-#include <cpphibernate/driver/mariadb/schema/table.h>
-#include <cpphibernate/driver/mariadb/schema/filter.h>
-
-using namespace ::cpphibernate::driver::mariadb_impl;
-
-bool filter_t::contains(const table_t* table, bool check_base) const
-{
-    if (tables.count(table))
-        return true;
-    else if (check_base && table->base_table)
-        return contains(table->base_table, true);
-    else
-        return false;
-}
-
-bool filter_t::contains(const field_t* field) const
-{
-    return (fields.count(field) > 0);
-}
\ No newline at end of file
diff --git a/src/driver/mariadb/schema/table.cpp b/src/driver/mariadb/schema/table.cpp
index 4135e8a..6e48ecd 100644
--- a/src/driver/mariadb/schema/table.cpp
+++ b/src/driver/mariadb/schema/table.cpp
@@ -12,6 +12,173 @@
 using namespace ::utl;
 using namespace ::cpphibernate::driver::mariadb_impl;
 
+/* data_extractor_t */
+
+struct data_extractor_t
+{
+    const table_t&              _table;
+    const read_context&         _context;
+    const ::cppmariadb::row&    _row;
+
+    mutable size_t              _index;
+
+    data_extractor_t(
+        const table_t&              p_table,
+        const read_context&         p_context,
+        const ::cppmariadb::row&    p_row)
+        : _table        (p_table)
+        , _context      (p_context)
+        , _row          (p_row)
+        { }
+
+    inline value_t get_value() const
+    {
+        value_t ret;
+        auto f = _row.at(_index);
+        if (!f.is_null())
+            ret = f.get<std::string>();
+        return ret;
+    }
+
+    inline void read_field(const field_t& field) const
+    {
+        field.set(_context, get_value());
+        ++_index;
+    }
+
+    inline void read_table(const table_t& table) const
+    {
+        if (table.base_table)
+            read_table(*table.base_table);
+
+        if (_context.filter.is_excluded(table))
+            return;
+
+        /* primary key */
+        assert(table.primary_key_field);
+        read_field(*table.primary_key_field);
+
+        /* data fields */
+        for (auto& ptr : table.data_fields)
+        {
+            assert(ptr);
+            auto& field = *ptr;
+            if (!_context.filter.is_excluded(field))
+                read_field(field);
+        }
+    }
+
+    inline void operator()() const
+    {
+        _index = 0;
+        _context.emplace();
+        read_table(_table);
+    }
+};
+
+/* select_query_builder_t */
+
+struct select_query_builder_t
+{
+    const table_t&      _table;
+    const filter_t&     _filter;
+    bool                _is_dynamic;
+
+    size_t              index { 0 };
+    std::ostringstream  os;
+    std::ostringstream  join;
+
+    select_query_builder_t(
+        const table_t&  p_table,
+        const filter_t& p_filter,
+        bool            p_is_dynamic)
+        : _table     (p_table)
+        , _filter    (p_filter)
+        , _is_dynamic(p_is_dynamic)
+        { }
+
+    inline void add_field(const field_t& field)
+    {
+        if (index++) os << ", ";
+        os  <<  "`"
+            <<  field.table_name
+            <<  "`.`"
+            <<  field.field_name
+            <<  "`";
+    }
+
+    inline bool add_table(const table_t& table, const std::string& prefix)
+    {
+        bool ret = false;
+
+        if (table.base_table)
+        {
+            auto tmp = add_table(*table.base_table, "");
+            if (tmp)
+            {
+                assert(table.base_table->primary_key_field);
+                auto& base_key = *table.base_table->primary_key_field;
+                ret = true;
+                join    <<  " LEFT JOIN `"
+                        <<  table.table_name
+                        <<  "` ON `"
+                        <<  table.table_name
+                        <<  "`.`"
+                        <<  base_key.field_name
+                        <<  "`=`"
+                        <<  base_key.table_name
+                        <<  "`.`"
+                        <<  base_key.field_name
+                        <<  "`";
+            }
+        }
+
+        /* __type */
+        if (    _is_dynamic
+            && !table.base_table
+            && !table.derived_tables.empty())
+        {
+            if (index++) os << ", ";
+            os  <<  "`"
+                <<  table.table_name
+                <<  "`.`__type` AS `__type`";
+            ret = true;
+        }
+
+        if (_filter.is_excluded(table))
+            return ret;
+
+        ret = true;
+
+        /* primary key */
+        assert(table.primary_key_field);
+        add_field(*table.primary_key_field);
+
+        /* data fields */
+        for (auto& ptr : table.data_fields)
+        {
+            assert(ptr);
+            auto& field = *ptr;
+            if (!_filter.is_excluded(field))
+                add_field(field);
+        }
+
+        return ret;
+    }
+
+    inline std::string operator()()
+    {
+        os  <<  "SELECT ";
+        add_table(_table, "");
+        os  <<  " FROM `"
+            <<  _table.table_name
+            <<  "`"
+            <<  join.str()
+            <<  " ?where! ?order! ?limit!";
+        return os.str();
+    }
+};
+
 /* build queries */
 
 std::string build_init_stage1_query(const table_t& table)
@@ -380,7 +547,7 @@ std::string build_create_update_query(const table_t& table, const filter_t* filt
     /* base table key fields */
     if (    static_cast<bool>(table.base_table)
         && (   !is_update
-            ||  filter->contains(table.base_table, true)))
+            || !filter->is_excluded(*table.base_table)))
     {
         if (index++)
             os << ", ";
@@ -401,11 +568,11 @@ std::string build_create_update_query(const table_t& table, const filter_t* filt
     for (auto& ptr : table.foreign_table_one_fields)
     {
         assert(static_cast<bool>(ptr));
-        if (is_update && !filter->contains(ptr))
+        auto& field_info = *ptr;
+        if (is_update && filter->is_excluded(field_info))
             continue;
         if (index++)
             os << ", ";
-        auto&  field_info = *ptr;
         assert(field_info.referenced_table);
         assert(field_info.referenced_table->primary_key_field);
         auto&  key_info = *field_info.referenced_table->primary_key_field;
@@ -462,12 +629,12 @@ std::string build_create_update_query(const table_t& table, const filter_t* filt
     /* data fields */
     for (auto& ptr : table.data_fields)
     {
-        if (is_update && !filter->contains(ptr))
+        assert(ptr);
+        auto& field_info = *ptr;
+        if (is_update && filter->is_excluded(field_info))
             continue;
         if (index++)
             os << ", ";
-        assert(static_cast<bool>(ptr));
-        auto&  field_info = *ptr;
         os  <<  "`"
             <<  field_info.field_name
             <<  "`="
@@ -505,6 +672,14 @@ std::string build_create_update_query(const table_t& table, const filter_t* filt
     return os.str();
 }
 
+std::string build_select_query(
+    const table_t& table,
+    const filter_t& filter,
+    bool is_dynamic)
+{
+    return select_query_builder_t(table, filter, is_dynamic)();
+}
+
 /* execute_create_update */
 
 std::string table_t::execute_create_update(
@@ -537,7 +712,7 @@ std::string table_t::execute_create_update(
     /* base_key */
     if (    base_table
         && (   !is_update
-            ||  filter->contains(base_table, true)))
+            || !filter->is_excluded(*base_table)))
     {
         auto new_context = context;
         if (!new_context.derived_table)
@@ -547,14 +722,14 @@ std::string table_t::execute_create_update(
         ++index;
     }
 
-    if (is_update && !filter->contains(this, false))
+    if (is_update && filter->is_excluded(*this))
         return primary_key;
 
     /* foreign table one fields */
     for (auto& ptr : foreign_table_one_fields)
     {
         assert(ptr);
-        if (is_update && !filter->contains(ptr))
+        if (is_update && filter->is_excluded(*ptr))
             continue;
         value_t key = ptr->foreign_create_update(context);
         if (key.has_value())    statement.set(index, std::move(key));
@@ -596,9 +771,9 @@ std::string table_t::execute_create_update(
     /* data fields */
     for (auto& ptr : data_fields)
     {
-        if (is_update && !filter->contains(ptr))
-            continue;
         assert(ptr);
+        if (is_update && filter->is_excluded(*ptr))
+            continue;
 
         auto& field_info = *ptr;
         auto  value      = field_info.get(context);
@@ -654,8 +829,8 @@ std::string table_t::execute_create_update(
     {
         assert(ptr);
         if (    is_update
-            && (   !filter->contains(ptr)
-                || !filter->contains(ptr->referenced_table, true)))
+            && (    filter->is_excluded(*ptr)
+                ||  filter->is_excluded(*ptr->referenced_table)))
             continue;
 
         auto next_context = context;
@@ -750,6 +925,20 @@ const table_t* table_t::get_derived(size_t id) const
     return *_statement_create_table;
 }
 
+::cppmariadb::statement& table_t::get_statement_select(const read_context& context) const
+{
+    auto& map = context.is_dynamic
+        ? _statement_select_dynamic
+        : _statement_select_static;
+    auto it = map.find(context.filter.cache_id);
+    if (it == map.end())
+    {
+        auto query = build_select_query(*this, context.filter, context.is_dynamic);
+        it = map.emplace(context.filter.cache_id, ::cppmariadb::statement(query)).first;
+    }
+    return it->second;
+}
+
 std::string table_t::create_update_base(const create_update_context& context) const
 {
     throw misc::hibernate_exception(static_cast<std::ostringstream&>(std::ostringstream { }
@@ -780,4 +969,19 @@ std::string table_t::create_update_exec(const create_update_context& context) co
 }
 
 std::string table_t::create_update_intern(const create_update_context& context) const
-    { return create_update_exec(context); }
\ No newline at end of file
+    { return create_update_exec(context); }
+
+void table_t::read_exec(const read_context& context) const
+{
+    auto& statement  = get_statement_select(context);
+    auto& connection = context.connection;
+    cpphibernate_debug_log("execute SELECT query: " << statement.query(connection));
+    auto result = connection.execute_used(statement);
+    if (!result)
+        throw misc::hibernate_exception("Unable to fetching data from database!");
+
+    ::cppmariadb::row* row;
+    while ((row = result->next()))
+        data_extractor_t(*this, context, *row)();
+    context.finish();
+}
\ No newline at end of file
diff --git a/test/cpphibernate_read.cpp b/test/cpphibernate_read.cpp
index 5aa523e..eadfb7c 100644
--- a/test/cpphibernate_read.cpp
+++ b/test/cpphibernate_read.cpp
@@ -6,13 +6,27 @@
 
 using namespace ::testing;
 using namespace ::cpphibernate;
+using namespace ::cpphibernate::modifier;
+using namespace ::boost::hana::literals;
 
 TEST(CppHibernateTests, read_test1)
 {
     StrictMock<mariadb_mock> mock;
 
-    // expect_query(mock,  "START TRANSACTION");
-    // expect_query(mock,  "COMMIT");
+    expect_query(mock,  "START TRANSACTION");
+    expect_query(mock,  "SELECT "
+                            "`tbl_test1`.`tbl_test1_id`, "
+                            "`tbl_test1`.`str_data`, "
+                            "`tbl_test1`.`str64_data`, "
+                            "`tbl_test1`.`u32_nullable`, "
+                            "`tbl_test1`.`u32_ptr_u`, "
+                            "`tbl_test1`.`u32_ptr_s` "
+                        "FROM "
+                            "`tbl_test1`   ",
+                        result_used({
+                           { "3d12697a-abb9-11e8-98d0-529269fb1459", "str_data of class `test1` object `t1`", "str64_data of class `test1` object `t1`", nullptr, "123", "456" }
+                        }));
+    expect_query(mock,  "COMMIT");
 
     EXPECT_CALL(
         mock,
@@ -26,9 +40,105 @@ TEST(CppHibernateTests, read_test1)
             .WillRepeatedly(WithArgs<1, 2, 3>(EscapeString()));
 
     test1 t1;
-    t1.id = uuid("1e133ad8-ad2e-11e8-98d0-529269fb1459");
+    t1.id = uuid("3d12697a-abb9-11e8-98d0-529269fb1459");
 
     ::cppmariadb::connection connection(reinterpret_cast<MYSQL*>(0x1111));
     auto context = make_context<driver::mariadb>(test_schema, connection);
+
+    using namespace modifier;
     context.read(t1);
-}
\ No newline at end of file
+
+    EXPECT_EQ   (t1.str_data,   "str_data of class `test1` object `t1`");
+    EXPECT_EQ   (t1.str64_data, "str64_data of class `test1` object `t1`");
+    EXPECT_FALSE(static_cast<bool>(t1.u32_nullable));
+    ASSERT_TRUE (static_cast<bool>(t1.u32_ptr_u));
+    EXPECT_EQ   (*t1.u32_ptr_u, 123);
+    ASSERT_TRUE (static_cast<bool>(t1.u32_ptr_s));
+    EXPECT_EQ   (*t1.u32_ptr_s, 456);
+}
+
+TEST(CppHibernateTests, read_test2)
+{
+    StrictMock<mariadb_mock> mock;
+
+    expect_query(mock,  "START TRANSACTION");
+    expect_query(mock,  "SELECT "
+                            "`tbl_test2`.`tbl_test2_id`, "
+                            "`tbl_test2`.`u8_data`, "
+                            "`tbl_test2`.`i8_data`, "
+                            "`tbl_test2`.`u16_data`, "
+                            "`tbl_test2`.`i16_data` "
+                        "FROM "
+                            "`tbl_test2`   ",
+                        result_used({
+                            { "3d1270dc-abb9-11e8-98d0-529269fb1459", "1", "2", "3", "4" }
+                        }));
+    expect_query(mock,  "COMMIT");
+
+    EXPECT_CALL(
+        mock,
+        mysql_real_escape_string(reinterpret_cast<MYSQL*>(0x1111), _, _, _))
+            .Times(AnyNumber())
+            .WillRepeatedly(WithArgs<1, 2, 3>(EscapeString()));
+
+    EXPECT_CALL(
+        mock,
+        mysql_close(
+            reinterpret_cast<MYSQL*>(0x1111)));
+
+    ::cppmariadb::connection connection(reinterpret_cast<MYSQL*>(0x1111));
+    auto context = make_context<driver::mariadb>(test_schema, connection);
+
+    constexpr decltype(auto) test2_key_field = test_schema.tables[1_c].fields[0_c];
+
+    test2 t2;
+    context.read(t2, where(equal(test2_key_field, "3d1270dc-abb9-11e8-98d0-529269fb1459")));
+
+    EXPECT_EQ(1, t2.u8_data);
+    EXPECT_EQ(2, t2.i8_data);
+    EXPECT_EQ(3, t2.u16_data);
+    EXPECT_EQ(4, t2.i16_data);
+}
+
+TEST(CppHibernateTests, read_test3)
+{
+    StrictMock<mariadb_mock> mock;
+
+    expect_query(mock,  "START TRANSACTION");
+    expect_query(mock,  "SELECT "
+                            "`tbl_test3`.`tbl_test3_id`, "
+                            "`tbl_test3`.`u32_data`, "
+                            "`tbl_test3`.`i32_data`, "
+                            "`tbl_test3`.`u64_data`, "
+                            "`tbl_test3`.`i64_data` "
+                        "FROM "
+                            "`tbl_test3`   ",
+                        result_used({
+                            { "3d12737a-abb9-11e8-98d0-529269fb1459", "5", "6", "7", "8" }
+                        }));
+    expect_query(mock,  "COMMIT");
+
+    EXPECT_CALL(
+        mock,
+        mysql_real_escape_string(reinterpret_cast<MYSQL*>(0x1111), _, _, _))
+            .Times(AnyNumber())
+            .WillRepeatedly(WithArgs<1, 2, 3>(EscapeString()));
+
+    EXPECT_CALL(
+        mock,
+        mysql_close(
+            reinterpret_cast<MYSQL*>(0x1111)));
+
+    ::cppmariadb::connection connection(reinterpret_cast<MYSQL*>(0x1111));
+    auto context = make_context<driver::mariadb>(test_schema, connection);
+
+    test3 t3;
+    t3.id = uuid("3d12737a-abb9-11e8-98d0-529269fb1459");
+
+    context.read(t3);
+
+    EXPECT_EQ(5, t3.u32_data);
+    EXPECT_EQ(6, t3.i32_data);
+    EXPECT_EQ(7, t3.u64_data);
+    EXPECT_EQ(8, t3.i64_data);
+}
diff --git a/test/test_helper.h b/test/test_helper.h
index 4ab3b54..420b2c2 100644
--- a/test/test_helper.h
+++ b/test/test_helper.h
@@ -24,7 +24,7 @@ ACTION(EscapeString)
 struct result_data
     : public mariadb_mock_item
 {
-    using row_type  = std::vector<std::string>;
+    using row_type  = std::vector<const char *>;
     using data_type = std::vector<row_type>;
 
     struct internal_data_t
@@ -55,8 +55,8 @@ struct result_data
             for (size_t j = 0; j < d.size(); ++j)
             {
                 auto& str = d.at(j);
-                intern.data[j]   = const_cast<char*>(str.c_str());
-                intern.length[j] = static_cast<unsigned long>(str.size());
+                intern.data[j]   = const_cast<char*>(str);
+                intern.length[j] = static_cast<unsigned long>(str ? strlen(str) : 0);
             }
         }
     }