diff --git a/CHANGELOG b/CHANGELOG
index 523faff3e..411dfd930 100644
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,5 +1,7 @@
 === master
 
+* Support json_table on PostgreSQL 17+ in the pg_json_ops extension (jeremyevans)
+
 * Make Dataset#get and #first without argument not create intermediate datasets if receiver uses raw SQL (jeremyevans)
 
 * Add dataset_run extension, for building SQL using datasets, and running with Database#run (jeremyevans)
diff --git a/lib/sequel/extensions/pg_json_ops.rb b/lib/sequel/extensions/pg_json_ops.rb
index b7d1855b3..24ee99753 100644
--- a/lib/sequel/extensions/pg_json_ops.rb
+++ b/lib/sequel/extensions/pg_json_ops.rb
@@ -88,6 +88,12 @@
 #   j.path_query_array('$.foo') # jsonb_path_query_array(jsonb_column, '$.foo')
 #   j.path_query_first('$.foo') # jsonb_path_query_first(jsonb_column, '$.foo')
 #
+# For the PostgreSQL 12+ SQL/JSON path functions, one argument is required (+path+) and
+# two more arguments are optional (+vars+ and +silent+).  +path+ specifies the JSON path.
+# +vars+ specifies a hash or a string in JSON format of named variables to be
+# substituted in +path+. +silent+ specifies whether errors are suppressed. By default,
+# errors are not suppressed.
+#
 # On PostgreSQL 13+ timezone-aware SQL/JSON path functions and operators are supported:
 #
 #   j.path_exists_tz!('$.foo')     # jsonb_path_exists_tz(jsonb_column, '$.foo')
@@ -96,12 +102,6 @@
 #   j.path_query_array_tz('$.foo') # jsonb_path_query_array_tz(jsonb_column, '$.foo')
 #   j.path_query_first_tz('$.foo') # jsonb_path_query_first_tz(jsonb_column, '$.foo')
 #
-# For the PostgreSQL 12+ SQL/JSON path functions, one argument is required (+path+) and
-# two more arguments are optional (+vars+ and +silent+).  +path+ specifies the JSON path.
-# +vars+ specifies a hash or a string in JSON format of named variables to be
-# substituted in +path+. +silent+ specifies whether errors are suppressed. By default,
-# errors are not suppressed.
-#
 # On PostgreSQL 14+, The JSONB <tt>[]</tt> method will use subscripts instead of being
 # the same as +get+, if the value being wrapped is an identifer:
 #
@@ -129,8 +129,8 @@
 #   j.is_json(type: :object)               # j IS JSON OBJECT
 #   j.is_json(type: :object, unique: true) # j IS JSON OBJECT WITH UNIQUE
 #   j.is_not_json                          # j IS NOT JSON
-#   j.is_not_json(type: :array)                # j IS NOT JSON ARRAY
-#   j.is_not_json(unique: true)                # j IS NOT JSON WITH UNIQUE
+#   j.is_not_json(type: :array)            # j IS NOT JSON ARRAY
+#   j.is_not_json(unique: true)            # j IS NOT JSON WITH UNIQUE
 #
 # On PostgreSQL 17+, the additional JSON functions are supported (see method documentation
 # for additional options):
@@ -143,6 +143,29 @@
 #   j.value('$.foo', returning: Time)  # json_value(jsonb_column, '$.foo' RETURNING timestamp)
 #   j.query('$.foo', wrapper: true)    # json_query(jsonb_column, '$.foo' WITH WRAPPER)
 #
+#   j.table('$.foo') do
+#      String :bar
+#      Integer :baz
+#   end
+#   # json_table('$.foo' COLUMNS(bar text, baz integer))
+#
+#   j.table('$.foo', passing: {a: 1}) do
+#      ordinality :id
+#      String :bar, format: :json, on_error: :empty_object
+#      nested '$.baz' do
+#        Integer :q, path: '$.quux', on_empty: :error
+#      end
+#      exists :x, on_error: false
+#   end
+#   # json_table("j", '$.foo' PASSING 1 AS a COLUMNS(
+#   #   "id" FOR ORDINALITY,
+#   #   "bar" text FORMAT JSON EMPTY OBJECT ON ERROR,
+#   #   NESTED '$.baz' COLUMNS(
+#   #     "q" integer PATH '$.quux' ERROR ON EMPTY
+#   #   ),
+#   #   "d" date EXISTS FALSE ON ERROR
+#   # ))
+#
 # If you are also using the pg_json extension, you should load it before
 # loading this extension.  Doing so will allow you to use the #op method on
 # JSONHash, JSONHarray, JSONBHash, and JSONBArray, allowing you to perform json/jsonb operations
@@ -364,6 +387,72 @@ def strip_nulls
         self.class.new(function(:strip_nulls))
       end
 
+      # Returns json_table SQL function expression, querying JSON data and returning
+      # the results as a relational view, which can be accessed similarly to a regular
+      # SQL table. This accepts a block that is handled in a similar manner to
+      # Database#create_table, though it operates differently.
+      #
+      # Table level options:
+      #
+      # :on_error :: How to handle errors when evaluating the JSON path expression.
+      #              :empty_array :: Return an empty array/result set
+      #              :error :: raise a DatabaseError
+      # :passing :: Variables to pass to the JSON path expression.  Keys are variable
+      #             names, values are the values of the variable.
+      #
+      # Inside the block, the following methods can be used:
+      #
+      # ordinality(name) :: Include a FOR ORDINALITY column, which operates similar to an
+      #                     autoincrementing primary key.
+      # column(name, type, opts={}) :: Return a normal column that uses the given type.
+      # exists(name, type, opts={}) :: Return a boolean column for whether the JSON path yields any values.
+      # nested(path, &block) :: Extract nested data from the result set at the given path.
+      #                         This block is treated the same as a json_table block, and
+      #                         arbitrary levels of nesting are supported.
+      # 
+      # The +column+ method supports the following options:
+      # 
+      # :path :: JSON path to the object (the default is <tt>$.NAME</tt>, where +NAME+ is the
+      #          name of the column).
+      # :format :: Set to +:json+ to use FORMAT JSON, when you expect the value to be a
+      #            valid JSON object.
+      # :on_empty, :on_error :: How to handle case where JSON path evaluation is empty or
+      #                         results in an error. Values supported are:
+      #                         :empty_array :: Return empty array (requires <tt>format: :json</tt>)
+      #                         :empty_object :: Return empty object (requires <tt>format: :json</tt>)
+      #                         :error :: Raise a DatabaseError
+      #                         :null :: Return nil (NULL)
+      # :wrapper :: How to wrap returned values:
+      #             true, :unconditional :: Always wrap returning values in an array
+      #             :conditional :: Only wrap multiple return values in an array
+      #             :keep_quotes :: Wrap scalar strings in quotes
+      #             :omit_quotes :: Do not wrap scalar strings in quotes
+      #
+      # The +exists+ method supports the following options:
+      #
+      # :path :: JSON path to the object (same as +column+ option)
+      # :on_error :: How to handle case where JSON path evaluation results in an error.
+      #              Values supported are:
+      #              :error :: Raise a DatabaseError
+      #              true :: Return true
+      #              false :: Return false
+      #              :null :: Return nil (NULL)
+      #
+      # Inside the block, methods for Ruby class names are also supported, allowing you
+      # to use syntax such as:
+      #
+      #   json_op.table('$.a') do
+      #     String :b
+      #     Integer :c, path: '$.d'
+      #   end
+      #
+      # One difference between this method and Database#create_table is that method_missing
+      # is not supported inside the block.  Use the +column+ method for PostgreSQL types
+      # that are not mapped to Ruby classes.
+      def table(path, opts=OPTS, &block)
+        JSONTableOp.new(self, path, opts, &block)
+      end
+
       # Builds arbitrary record from json object.  You need to define the
       # structure of the record using #as on the resulting object:
       #
@@ -1032,6 +1121,223 @@ def on_sql_value(value)
       end
     end
 
+    # Object representing json_table calls
+    class JSONTableOp < SQL::Expression
+      TABLE_ON_ERROR_SQL = {
+        :error => ' ERROR ON ERROR',
+        :empty_array => ' EMPTY ARRAY ON ERROR',
+      }.freeze
+      private_constant :TABLE_ON_ERROR_SQL
+
+      COLUMN_ON_SQL = {
+        :null =>  ' NULL',
+        :error => ' ERROR',
+        :empty_array => ' EMPTY ARRAY',
+        :empty_object => ' EMPTY OBJECT',
+      }.freeze
+      private_constant :COLUMN_ON_SQL
+
+      EXISTS_ON_ERROR_SQL = {
+        :error => ' ERROR',
+        true => ' TRUE',
+        false => ' FALSE',
+        :null => ' UNKNOWN',
+      }.freeze
+      private_constant :EXISTS_ON_ERROR_SQL
+
+      WRAPPER = {
+        :conditional => ' WITH CONDITIONAL WRAPPER',
+        :unconditional => ' WITH WRAPPER',
+        :omit_quotes => ' OMIT QUOTES',
+        :keep_quotes => ' KEEP QUOTES',
+      }
+      WRAPPER[true] = WRAPPER[:unconditional]
+      WRAPPER.freeze
+      private_constant :WRAPPER
+
+      # Class used to evaluate json_table blocks and nested blocks
+      class ColumnDSL
+        # Return array of column information recorded for the instance
+        attr_reader :columns
+
+        def self.columns(&block)
+          new(&block).columns.freeze
+        end
+
+        def initialize(&block)
+          @columns = []
+          instance_exec(&block)
+        end
+
+        # Include a FOR ORDINALITY column
+        def ordinality(name)
+          @columns << [:ordinality, name].freeze
+        end
+
+        # Include a regular column with the given type
+        def column(name, type, opts=OPTS)
+          @columns << [:column, name, type, opts].freeze
+        end
+
+        # Include an EXISTS column with the given type
+        def exists(name, type, opts=OPTS)
+          @columns << [:exists, name, type, opts].freeze
+        end
+
+        # Include a nested set of columns at the given path.
+        def nested(path, &block)
+          @columns << [:nested, path, ColumnDSL.columns(&block)].freeze
+        end
+
+        # Include a bigint column
+        def Bignum(name, opts=OPTS)
+          @columns << [:column, name, :Bignum, opts].freeze
+        end
+
+        # Define methods for handling other generic types
+        %w'String Integer Float Numeric BigDecimal Date DateTime Time File TrueClass FalseClass'.each do |meth|
+          klass = Object.const_get(meth)
+          define_method(meth) do |name, opts=OPTS|
+            @columns << [:column, name, klass, opts].freeze
+          end
+        end
+      end
+      private_constant :ColumnDSL
+
+      # See JSONBaseOp#table for documentation on the options.
+      def initialize(expr, path, opts=OPTS, &block)
+        @expr = expr
+        @path = path
+        @passing = opts[:passing]
+        @on_error = opts[:on_error]
+        @columns = opts[:_columns] || ColumnDSL.columns(&block)
+        freeze
+      end
+
+      # Append the json_table function call expression to the SQL
+      def to_s_append(ds, sql)
+        sql << 'json_table('
+        ds.literal_append(sql, @expr)
+        sql << ', '
+        default_literal_append(ds, sql, @path)
+
+        if (passing = @passing) && !passing.empty?
+          sql << ' PASSING '
+          comma = false
+          passing.each do |k, v|
+            if comma
+              sql << ', '
+            else
+              comma = true
+            end
+            ds.literal_append(sql, v)
+            sql << " AS " << k.to_s
+          end
+        end
+
+        to_s_append_columns(ds, sql, @columns)
+        sql << TABLE_ON_ERROR_SQL.fetch(@on_error) if @on_error
+        sql << ')'
+      end
+
+      # Support transforming of json_table expression
+      def sequel_ast_transform(transformer)
+        opts = {:on_error=>@on_error, :_columns=>@columns}
+
+        if @passing
+          passing = opts[:passing] = {}
+          @passing.each do |k, v|
+            passing[k] = transformer.call(v)
+          end
+        end
+
+        self.class.new(transformer.call(@expr), @path, opts)
+      end
+
+      private
+
+      # Append the set of column information to the SQL.  Separated to handle
+      # nested sets of columns.
+      def to_s_append_columns(ds, sql, columns)
+        sql << ' COLUMNS('
+        comma = nil
+        columns.each do |column|
+          if comma
+            sql << comma
+          else
+            comma = ', '
+          end
+          to_s_append_column(ds, sql, column)
+        end
+        sql << ')'
+      end
+
+      # Append the column information to the SQL.  Handles the various
+      # types of json_table columns.
+      def to_s_append_column(ds, sql, column)
+        case column[0]
+        when :column
+          _, name, type, opts = column
+          ds.literal_append(sql, name)
+          sql << ' ' << ds.db.send(:type_literal, opts.merge(:type=>type)).to_s
+          sql << ' FORMAT JSON' if opts[:format] == :json
+          to_s_append_path(ds, sql, opts[:path])
+          sql << WRAPPER.fetch(opts[:wrapper]) if opts[:wrapper]
+          to_s_append_on_value(ds, sql, opts[:on_empty], " ON EMPTY")
+          to_s_append_on_value(ds, sql, opts[:on_error], " ON ERROR")
+        when :ordinality
+          ds.literal_append(sql, column[1])
+          sql << ' FOR ORDINALITY'
+        when :exists
+          _, name, type, opts = column
+          ds.literal_append(sql, name)
+          sql << ' ' << ds.db.send(:type_literal, opts.merge(:type=>type)).to_s
+          sql << ' EXISTS'
+          to_s_append_path(ds, sql, opts[:path])
+          unless (on_error = opts[:on_error]).nil?
+            sql << EXISTS_ON_ERROR_SQL.fetch(on_error) << " ON ERROR"
+          end
+        else # when :nested
+          _, path, columns = column
+          sql << 'NESTED '
+          default_literal_append(ds, sql, path)
+          to_s_append_columns(ds, sql, columns)
+        end
+      end
+
+      # Handle DEFAULT values in ON EMPTY/ON ERROR fragments
+      def to_s_append_on_value(ds, sql, value, cond)
+        if value
+          if v = COLUMN_ON_SQL[value]
+            sql << v
+          else
+            sql << ' DEFAULT '
+            default_literal_append(ds, sql, value)
+          end
+          sql << cond
+        end
+      end
+
+      # Append path caluse to the SQL
+      def to_s_append_path(ds, sql, path)
+        if path
+          sql << ' PATH '
+          default_literal_append(ds, sql, path)
+        end
+      end
+
+      # Do not auto paramterize default value or path value, as PostgreSQL doesn't allow it.
+      def default_literal_append(ds, sql, v)
+        if sql.respond_to?(:skip_auto_param)
+          sql.skip_auto_param do
+            ds.literal_append(sql, v)
+          end
+        else
+          ds.literal_append(sql, v)
+        end
+      end
+    end
+
     module JSONOpMethods
       # Wrap the receiver in an JSONOp so you can easily use the PostgreSQL
       # json functions and operators with it.
diff --git a/spec/adapters/postgres_spec.rb b/spec/adapters/postgres_spec.rb
index 9110209f7..4d120aa61 100644
--- a/spec/adapters/postgres_spec.rb
+++ b/spec/adapters/postgres_spec.rb
@@ -4366,6 +4366,194 @@ def left_item_id
       @db.get(j.query('$.n[1]', :wrapper=>:conditional)).must_equal [3]
       @db.get(j.query('$.s', returning: String)).must_equal '"t"'
       @db.get(j.query('$.s', returning: String, :wrapper=>:omit_quotes)).must_equal "t"
+
+      j = Sequel.send(:"pg_#{json_type}_op", (<<JSON).gsub(/\s+/, ' '))
+{ "favorites" : [
+   { "kind" : "comedy", "films" : [
+     { "title" : "Bananas",
+       "director" : "Woody Allen"},
+     { "title" : "The Dinner Game",
+       "director" : "Francis Veber" } ] },
+   { "kind" : "horror", "films" : [
+     { "title" : "Psycho",
+       "director" : "Alfred Hitchcock" } ] },
+   { "kind" : "thriller", "films" : [
+     { "title" : "Vertigo",
+       "director" : "Alfred Hitchcock" } ] },
+   { "kind" : "drama", "films" : [
+     { "title" : "Yojimbo",
+       "director" : "Akira Kurosawa" } ] }
+  ] }
+JSON
+
+      @db.from(j.table('$.favorites[*]') do
+        ordinality :id
+        String :kind
+        String :title, path: '$.films[*].title', :wrapper=>true
+        String :director, path: '$.films[*].director', :wrapper=>true
+      end).all.must_equal [
+        {:id=>1, :kind=>"comedy", :title=>'["Bananas", "The Dinner Game"]', :director=>'["Woody Allen", "Francis Veber"]'},
+        {:id=>2, :kind=>"horror", :title=>'["Psycho"]', :director=>'["Alfred Hitchcock"]'},
+        {:id=>3, :kind=>"thriller", :title=>'["Vertigo"]', :director=>'["Alfred Hitchcock"]'},
+        {:id=>4, :kind=>"drama", :title=>'["Yojimbo"]', :director=>'["Akira Kurosawa"]'}
+      ]
+
+      @db.from(j.table('$.favorites[*] ? (@.films[*].director == $filter)', :passing=>{:filter=>'Alfred Hitchcock'}) do
+        ordinality :id
+        String :kind, path: '$.kind'
+        String :title, path: '$.films[*].title', :wrapper=>:omit_quotes, :format=>:json
+        String :director, path: '$.films[*].director', :wrapper=>:keep_quotes
+      end).all.must_equal [
+        {:id=>1, :kind=>"horror", :title=>'Psycho', :director=>'"Alfred Hitchcock"'},
+        {:id=>2, :kind=>"thriller", :title=>'Vertigo', :director=>'"Alfred Hitchcock"'},
+      ]
+
+      @db.from(j.table('$.favorites[*] ? (@.films[*].director == $filter)', :passing=>{:filter=>'Alfred Hitchcock'}) do
+        ordinality :id
+        String :kind, path: '$.kind'
+        nested '$.films[*]' do
+          String :title, path: '$.title', :wrapper=>:omit_quotes, :format=>:json
+          String :director, path: '$.director', :wrapper=>:keep_quotes
+        end
+      end).all.must_equal [
+        {:id=>1, :kind=>"horror", :title=>'Psycho', :director=>'"Alfred Hitchcock"'},
+        {:id=>2, :kind=>"thriller", :title=>'Vertigo', :director=>'"Alfred Hitchcock"'},
+      ]
+
+      @db.from(j.table('$.favorites[*]') do
+        ordinality :id
+        String :kind, path: '$.kind'
+        nested '$.films[*]' do
+          String :title, path: '$.title', :wrapper=>:omit_quotes, :format=>:json
+          String :director, path: '$.director', :wrapper=>:keep_quotes
+        end
+      end).all.must_equal [
+        {:id=>1, :kind=>"comedy", :title=>'Bananas', :director=>'"Woody Allen"'},
+        {:id=>1, :kind=>"comedy", :title=>'The Dinner Game', :director=>'"Francis Veber"'},
+        {:id=>2, :kind=>"horror", :title=>'Psycho', :director=>'"Alfred Hitchcock"'},
+        {:id=>3, :kind=>"thriller", :title=>'Vertigo', :director=>'"Alfred Hitchcock"'},
+        {:id=>4, :kind=>"drama", :title=>'Yojimbo', :director=>'"Akira Kurosawa"'}
+      ]
+
+      j = Sequel.send(:"pg_#{json_type}_op", (<<JSON).gsub(/\s+/, ' '))
+{"favorites":
+    {"movies":
+      [{"name": "One", "director": "John Doe"},
+       {"name": "Two", "director": "Don Joe"}],
+     "books":
+      [{"name": "Mystery", "authors": [{"name": "Brown Dan"}]},
+       {"name": "Wonder", "authors": [{"name": "Jun Murakami"}, {"name":"Craig Doe"}]}]
+}}
+JSON
+      @db.from(j.table('$.favorites[*]') do
+        ordinality :user_id
+        nested '$.movies[*]' do
+          ordinality :movie_id
+          String :mname, path: '$.name'
+          String :director
+        end
+        nested '$.books[*]' do
+          ordinality :book_id
+          String :bname, path: '$.name'
+          nested '$.authors[*]' do
+            ordinality :author_id
+            String :author_name, path: '$.name'
+          end
+        end
+      end).all.must_equal [
+        {:user_id=>1, :movie_id=>1, :mname=>"One", :director=>"John Doe", :book_id=>nil, :bname=>nil, :author_id=>nil, :author_name=>nil},
+        {:user_id=>1, :movie_id=>2, :mname=>"Two", :director=>"Don Joe", :book_id=>nil, :bname=>nil, :author_id=>nil, :author_name=>nil},
+        {:user_id=>1, :movie_id=>nil, :mname=>nil, :director=>nil, :book_id=>1, :bname=>"Mystery", :author_id=>1, :author_name=>"Brown Dan"},
+        {:user_id=>1, :movie_id=>nil, :mname=>nil, :director=>nil, :book_id=>2, :bname=>"Wonder", :author_id=>1, :author_name=>"Jun Murakami"},
+        {:user_id=>1, :movie_id=>nil, :mname=>nil, :director=>nil, :book_id=>2, :bname=>"Wonder", :author_id=>2, :author_name=>"Craig Doe"}
+      ]
+
+      proc do
+        @db.from(j.table('strict $.x[0].y', :on_error=>:error) do
+          ordinality :id
+        end).all
+      end.must_raise Sequel::DatabaseError
+
+      @db.from(j.table('strict $.x[0].y', :on_error=>:empty_array) do
+        ordinality :id
+      end).all.must_equal []
+
+      @db.from(j.table('strict $.x[0].y') do
+        ordinality :id
+      end).all.must_equal []
+
+      proc do
+        @db.from(j.table('$.favorites[*]') do
+          String :foo, path: 'strict $.foo', :on_error=>:error
+        end).all.must_equal []
+      end.must_raise Sequel::DatabaseError
+
+      @db.from(j.table('$.favorites[*]') do
+        String :foo, path: 'strict $.foo', :on_error=>:null
+      end).all.must_equal [{:foo=>nil}]
+
+      @db.from(j.table('$.favorites[*]') do
+        column :foo, json_type, path: 'strict $.foo', :on_error=>:empty_object, :format=>:json, :wrapper=>:conditional
+      end).all.must_equal [{:foo=>{}}]
+
+      @db.from(j.table('$.favorites[*]') do
+        column :foo, json_type, path: 'strict $.foo', :on_error=>:empty_array, :format=>:json
+      end).all.must_equal [{:foo=>[]}]
+
+      @db.from(j.table('$.favorites[*]') do
+        Integer :foo, path: 'strict $.foo', :on_error=>42
+      end).all.must_equal [{:foo=>42}]
+
+      proc do
+        @db.from(j.table('$.favorites[*]') do
+          String :foo, path: '$.foo', :on_empty=>:error
+        end).all.must_equal []
+      end.must_raise Sequel::DatabaseError
+
+      @db.from(j.table('$.favorites[*]') do
+        String :foo, path: '$.foo', :on_empty=>:null
+      end).all.must_equal [{:foo=>nil}]
+
+      @db.from(j.table('$.favorites[*]') do
+        column :foo, json_type, path: '$.foo', :on_empty=>:empty_object, :format=>:json
+      end).all.must_equal [{:foo=>{}}]
+
+      @db.from(j.table('$.favorites[*]') do
+        column :foo, json_type, path: '$.foo', :on_empty=>:empty_array, :format=>:json
+      end).all.must_equal [{:foo=>[]}]
+
+      @db.from(j.table('$.favorites[*]') do
+        Integer :foo, path: '$.foo', :on_empty=>42
+      end).all.must_equal [{:foo=>42}]
+
+      @db.from(j.table('$.favorites[*]') do
+        exists :foo, TrueClass
+      end).all.must_equal [{:foo=>false}]
+
+      @db.from(j.table('$.favorites[*]') do
+        exists :foo, TrueClass, path: 'strict $.foo', :on_error=>true
+      end).all.must_equal [{:foo=>true}]
+
+      @db.from(j.table('$.favorites[*]') do
+        exists :foo, TrueClass, path: 'strict $.foo', :on_error=>false
+      end).all.must_equal [{:foo=>false}]
+
+      @db.from(j.table('$.favorites[*]') do
+        exists :foo, String, path: 'strict $.foo', :on_error=>:null
+      end).all.must_equal [{:foo=>nil}]
+
+      proc do
+        @db.from(j.table('$.favorites[*]') do
+          exists :foo, String, path: 'strict $.foo', :on_error=>:error
+        end).all.must_equal [{:foo=>true}]
+      end.must_raise Sequel::DatabaseError
+
+      j = Sequel.send(:"pg_#{json_type}_op", '{"a":{"n": [{"b": 3}]}}')
+      @db.from(j.table('$.a'){column :j, json_type, :path=>'$.n[0]', :format=>:json, :wrapper=>true}).all.must_equal [{:j=>[{"b"=>3}]}]
+      @db.from(j.table('$.a'){column :j, json_type, :path=>'$.n[0]', :format=>:json, :wrapper=>:conditional}).all.must_equal [{:j=>{"b"=>3}}]
+
+      @db.from(j.table(Sequel['$.a.n'].as(:foo)){Integer :b}).all.must_equal [{:b=>3}]
+      @db.from(j.table(Sequel['$.a'].as(:foo)){nested(Sequel['$.n'].as(:c)){Integer :b}}).all.must_equal [{:b=>3}]
     end if DB.server_version >= 170000
   end
 end if DB.server_version >= 90200
diff --git a/spec/extensions/pg_json_ops_spec.rb b/spec/extensions/pg_json_ops_spec.rb
index 84633b133..181eedc21 100644
--- a/spec/extensions/pg_json_ops_spec.rb
+++ b/spec/extensions/pg_json_ops_spec.rb
@@ -621,4 +621,116 @@ def @db.server_version(*); 130000; end
   it "#path_query_first result should be a JSONBOp" do
     @l[@jb.path_query_first_tz('$').path_query_first_tz('$')].must_equal "jsonb_path_query_first_tz(jsonb_path_query_first_tz(j, '$'), '$')"
   end
+
+  it "#table should use the json_table function" do
+    @l[@j.table('$'){String :a}].must_equal "json_table(j, '$' COLUMNS(a text))"
+    @l[@j.table('$'){String :a; Integer :b}].must_equal "json_table(j, '$' COLUMNS(a text, b integer))"
+  end
+
+  it "#table should json_path_name via SQL::AliasedExpression" do
+    @l[@j.table(Sequel['$'].as(:foo)){String :a}].must_equal "json_table(j, '$' AS foo COLUMNS(a text))"
+  end
+
+  it "#table should support :passing option" do
+    @l[@j.table('$', :passing=>{}){String :a}].must_equal "json_table(j, '$' COLUMNS(a text))"
+    @l[@j.table('$', :passing=>{a: 1}){String :a}].must_equal "json_table(j, '$' PASSING 1 AS a COLUMNS(a text))"
+    @l[@j.table('$', :passing=>{a: 1, b: "2"}){String :a}].must_equal "json_table(j, '$' PASSING 1 AS a, '2' AS b COLUMNS(a text))"
+  end
+
+  it "#table should support :on_error option " do
+    @l[@j.table('$', :on_error=>:error){String :a}].must_equal "json_table(j, '$' COLUMNS(a text) ERROR ON ERROR)"
+    @l[@j.table('$', :on_error=>:empty_array){String :a}].must_equal "json_table(j, '$' COLUMNS(a text) EMPTY ARRAY ON ERROR)"
+  end
+
+  it "#table block should support #ordinality" do
+    @l[@j.table('$'){ordinality :a}].must_equal "json_table(j, '$' COLUMNS(a FOR ORDINALITY))"
+  end
+
+  it "#table block should support #exists" do
+    @l[@j.table('$'){exists :a, String}].must_equal "json_table(j, '$' COLUMNS(a text EXISTS))"
+  end
+
+  it "#table block #exists should support :path option" do
+    @l[@j.table('$'){exists :a, String, path: '$'}].must_equal "json_table(j, '$' COLUMNS(a text EXISTS PATH '$'))"
+  end
+
+  it "#table block #exists should support :on_error option" do
+    @l[@j.table('$'){exists :a, String, on_error: :error}].must_equal "json_table(j, '$' COLUMNS(a text EXISTS ERROR ON ERROR))"
+    @l[@j.table('$'){exists :a, String, on_error: true}].must_equal "json_table(j, '$' COLUMNS(a text EXISTS TRUE ON ERROR))"
+    @l[@j.table('$'){exists :a, String, on_error: false}].must_equal "json_table(j, '$' COLUMNS(a text EXISTS FALSE ON ERROR))"
+    @l[@j.table('$'){exists :a, String, on_error: :null}].must_equal "json_table(j, '$' COLUMNS(a text EXISTS UNKNOWN ON ERROR))"
+  end
+
+  it "#table block should support #column" do
+    @l[@j.table('$'){column :a, String}].must_equal "json_table(j, '$' COLUMNS(a text))"
+  end
+
+  it "#table block #column should support format: :json option" do
+    @l[@j.table('$'){column :a, String, format: :json}].must_equal "json_table(j, '$' COLUMNS(a text FORMAT JSON))"
+  end
+
+  it "#table block #column should support :path option" do
+    @l[@j.table('$'){column :a, String, path: '$'}].must_equal "json_table(j, '$' COLUMNS(a text PATH '$'))"
+  end
+
+  it "#table block #column should support :wrapper option" do
+    @l[@j.table('$'){column :a, String, :wrapper=>true}].must_equal "json_table(j, '$' COLUMNS(a text WITH WRAPPER))"
+    @l[@j.table('$'){column :a, String, :wrapper=>:conditional}].must_equal "json_table(j, '$' COLUMNS(a text WITH CONDITIONAL WRAPPER))"
+    @l[@j.table('$'){column :a, String, :wrapper=>:unconditional}].must_equal "json_table(j, '$' COLUMNS(a text WITH WRAPPER))"
+    @l[@j.table('$'){column :a, String, :wrapper=>:keep_quotes}].must_equal "json_table(j, '$' COLUMNS(a text KEEP QUOTES))"
+    @l[@j.table('$'){column :a, String, :wrapper=>:omit_quotes}].must_equal "json_table(j, '$' COLUMNS(a text OMIT QUOTES))"
+  end
+
+  it "#table block #column should support :on_empty option" do
+    @l[@j.table('$'){column :a, String, on_empty: :error}].must_equal "json_table(j, '$' COLUMNS(a text ERROR ON EMPTY))"
+    @l[@j.table('$'){column :a, String, on_empty: :null}].must_equal "json_table(j, '$' COLUMNS(a text NULL ON EMPTY))"
+    @l[@j.table('$'){column :a, String, on_empty: :empty_array, format: :json}].must_equal "json_table(j, '$' COLUMNS(a text FORMAT JSON EMPTY ARRAY ON EMPTY))"
+    @l[@j.table('$'){column :a, String, on_empty: :empty_object, format: :json}].must_equal "json_table(j, '$' COLUMNS(a text FORMAT JSON EMPTY OBJECT ON EMPTY))"
+    @l[@j.table('$'){column :a, String, on_empty: 42}].must_equal "json_table(j, '$' COLUMNS(a text DEFAULT 42 ON EMPTY))"
+  end
+
+  it "#table block #column should support :on_error option" do
+    @l[@j.table('$'){column :a, String, on_error: :error}].must_equal "json_table(j, '$' COLUMNS(a text ERROR ON ERROR))"
+    @l[@j.table('$'){column :a, String, on_error: :null}].must_equal "json_table(j, '$' COLUMNS(a text NULL ON ERROR))"
+    @l[@j.table('$'){column :a, String, on_error: :empty_array, format: :json}].must_equal "json_table(j, '$' COLUMNS(a text FORMAT JSON EMPTY ARRAY ON ERROR))"
+    @l[@j.table('$'){column :a, String, on_error: :empty_object, format: :json}].must_equal "json_table(j, '$' COLUMNS(a text FORMAT JSON EMPTY OBJECT ON ERROR))"
+    @l[@j.table('$'){column :a, String, on_error: 42}].must_equal "json_table(j, '$' COLUMNS(a text DEFAULT 42 ON ERROR))"
+  end
+
+  it "#table block should support #nested" do
+    @l[@j.table('$.a'){nested("$.b"){String :a}}].must_equal "json_table(j, '$.a' COLUMNS(NESTED '$.b' COLUMNS(a text)))"
+  end
+
+  it "#table block #nested should support json_path_name via SQL::AliasedExpression" do
+    @l[@j.table('$.a'){nested(Sequel["$.b"].as(:b)){String :a}}].must_equal "json_table(j, '$.a' COLUMNS(NESTED '$.b' AS b COLUMNS(a text)))"
+  end
+
+  it "#table block #nested should support arbitrary levels of nesting" do
+    @l[@j.table('$.a'){nested("$.b"){nested("$.c"){nested("$.d"){String :a}}}}].must_equal "json_table(j, '$.a' COLUMNS(NESTED '$.b' COLUMNS(NESTED '$.c' COLUMNS(NESTED '$.d' COLUMNS(a text)))))"
+  end
+
+  it "#table block should support Ruby class methods for generic types" do
+    @l[@j.table('$'){Bignum :a}].must_equal "json_table(j, '$' COLUMNS(a bigint))"
+    @l[@j.table('$'){Integer :a}].must_equal "json_table(j, '$' COLUMNS(a integer))"
+    @l[@j.table('$'){Float :a}].must_equal "json_table(j, '$' COLUMNS(a double precision))"
+    @l[@j.table('$'){Numeric :a}].must_equal "json_table(j, '$' COLUMNS(a numeric))"
+    @l[@j.table('$'){BigDecimal :a, :size=>[10, 2]}].must_equal "json_table(j, '$' COLUMNS(a numeric(10, 2)))"
+    @l[@j.table('$'){Date :a}].must_equal "json_table(j, '$' COLUMNS(a date))"
+    @l[@j.table('$'){DateTime :a}].must_equal "json_table(j, '$' COLUMNS(a timestamp))"
+    @l[@j.table('$'){Time :a}].must_equal "json_table(j, '$' COLUMNS(a timestamp))"
+    @l[@j.table('$'){Time :a, :only_time=>true}].must_equal "json_table(j, '$' COLUMNS(a time))"
+    @l[@j.table('$'){File :a}].must_equal "json_table(j, '$' COLUMNS(a bytea))"
+    @l[@j.table('$'){TrueClass :a}].must_equal "json_table(j, '$' COLUMNS(a boolean))"
+    @l[@j.table('$'){FalseClass :a}].must_equal "json_table(j, '$' COLUMNS(a boolean))"
+  end
+
+  it "#table should support AST transformations" do
+    @db.select(@j.table('$'){String :a}).qualify(:t).sql.must_equal "SELECT json_table(t.j, '$' COLUMNS(a text))"
+    @db.select(@j.table('$', :passing=>{a: :b}){String :a}).qualify(:t).sql.must_equal "SELECT json_table(t.j, '$' PASSING t.b AS a COLUMNS(a text))"
+  end
+
+  it "#table should not parameterize :on_empty/:on_error default option or any path options" do
+    @db.extension :pg_auto_parameterize
+    @db.select(@jb.table('$.a'){nested('$.b'){String :a, :path=>'$.c', on_empty: 1, on_error: 2}}).sql.must_equal "SELECT json_table(j, '$.a' COLUMNS(NESTED '$.b' COLUMNS(a text PATH '$.c' DEFAULT 1 ON EMPTY DEFAULT 2 ON ERROR)))"
+  end
 end